其他分享
首页 > 其他分享> > Flink之ProcessFunction的使用(1):定时器和状态管理的使用

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()
    }
}

 

标签:定时器,ProcessFunction,String,val,Flink,ctx,param,Long,温度
来源: https://www.cnblogs.com/yangshibiao/p/14133507.html