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

2020-12-14 15:32:31 阅读：186 来源： 互联网

标签：定时器 ProcessFunction String val Flink ctx param Long 温度

具体实现代码如下所示：

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

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

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