Flink之ProcessFunction的使用（1）：定时器和状态管理的使用

相关文章链接

具体实现代码如下所示：

main函数中代码如下：

val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

val socketStream: DataStream[String] = env.socketTextStream("localhost", 9999)

val sensorStream: DataStream[SensorReading] = socketStream.map(new MyMapToSensorReading)

// 检测每一个传感器的温度是否在10秒内连续上升（连续10秒上升后报警，并之后每一秒都报警，当温度下降时重新计时）
sensorStream.keyBy(_.id).process(new TempIncreWarning(10 * 1000))


env.execute("TimerAndStateDemo")

自定义类实现KeyedProcessFunction接口（KeyedProcessFunction接口为ProcessFunction接口的子接口）：

/**
 * 自定义的温度持续上升警报类
 * @param duration 温度持续上升时间
 */
class TempIncreWarning(duration:Long) extends KeyedProcessFunction[String, SensorReading, String] {

    /**
     * 知识点：
     *  1、ProcessFunction中的状态跟类的成员变量类似，但需要从环境上下文中获取，即运行了才有状态
     *  2、在方法中使用状态跟使用成员变量一样，但状态相对成员变量来说，安全性高、不会跟其他key冲突，一般使用状态
     */
    // 因为需要跟上一个温度做对比，将上一个温度封装到状态中（状态可以通过运行时上下文获取，也因为状态是在运行时上下文中，所以定义成lazy状态，防止编译时报错）
    lazy val lastTempState: ValueState[Double] =getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))
    // 为了方便删除定时器，还需要保存定时器的时间戳
    lazy val curTimerTsState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("cur-timer-ts", classOf[Long]))

    /**
     * 出来流中元素的方法，流中每一条元素都会调用一次此方法
     * @param value 流中的元素
     * @param ctx 环境上下文
     * @param out 将结果输出的类
     */
    override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {
        // 1、先将状态中的数据取出
        val lastTemp: Double = lastTempState.value()
        val curTimerTs: Long = curTimerTsState.value()

        // 2、更新温度
        lastTempState.update(value.temperature)

        // 3、判断温度变化，并采取对应操作
        if (value.temperature > lastTemp && curTimerTs == 0){
            // 3.1、当现在温度比上一个温度高，并且没有定时器时，注册一个定时器
            val ts: Long = ctx.timerService().currentProcessingTime() + duration
            ctx.timerService().registerProcessingTimeTimer(ts)
            curTimerTsState.update(ts)
        } else if (value.temperature < lastTemp){
            // 3.2、如果温度下降，删除定时器
            ctx.timerService().deleteProcessingTimeTimer(curTimerTs)
            curTimerTsState.clear()
        }
        // 3.3、当温度没变化时，不做操作
    }

    /**
     * 定时器触发时执行的操作（说明10s内没有温度值下降，报警）
     * @param timestamp 时间戳
     * @param ctx 环境上下文
     * @param out 输出类
     */
    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
        val key: String = ctx.getCurrentKey
        out.collect(s"传感器：$key, 已连续" + duration/1000 + "秒温度持续上升")
        curTimerTsState.clear()
    }
}