大数据之Flink(7) | KeyedProcessFunction

Bertha 。 2022-12-17 04:26 208阅读 0赞

需求：检测每一个传感器是否连续上升，在10秒钟内

import learning.SensorReading
    import org.apache.flink.api.common.state.{ ValueState, ValueStateDescriptor}
    import org.apache.flink.api.java.tuple.Tuple
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.util.Collector
    
    object ProcessFunctionTest { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
    
        val inputStream = env.socketTextStream("hadoop12",9999)
        val dataStream = inputStream
          .map(data => { 
            val dataArray = data.split(",")
            SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
          })
    
        //检测每一个传感器是否连续上升，在10秒钟内
        val warningStream = dataStream
          .keyBy("id")
          .process(new TempIncreWarning(10000L))
    
        warningStream.print()
        env.execute("process function job")
    
      }
    }
    //自定义KeyedProcessFunction Tuple只要是keyby中"id"
    class TempIncreWarning(interval: Long) extends KeyedProcessFunction[Tuple,SensorReading,String]{ 
      //由于需要跟之前的温度值做对比，所以将上一个温度保存成状态
      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]))
    
      override def processElement(value: SensorReading, ctx: KeyedProcessFunction[Tuple, SensorReading, String]
        #Context, out: Collector[String]): Unit = { 
        //首先取出状态
        val lastTemp = lastTempState.value()
        val curTimerTs = curTimerTsState.value()
    
        //将上次温度值得状态更新为当前数据的温度值
        lastTempState.update(value.temperature)
    
        //判断当前温度值，如果比之前温度高，并且没有定时器的话，就注册10s后的定时器
        //curTimerTs == 0就没有定时器
        if (value.temperature > lastTemp && curTimerTs == 0){ 
          val ts = ctx.timerService().currentProcessingTime() + interval
          ctx.timerService().registerProcessingTimeTimer(ts)
          curTimerTsState.update(ts)
        }
        //如果温度下降，删除定时器
        else if (value.temperature < lastTemp){ 
          ctx.timerService().deleteProcessingTimeTimer(curTimerTs)
          //清空状态
          curTimerTsState.clear()
        }
      }
    
      //定时器触发，说明10秒内没来下降的温度值，报警
      override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Tuple, SensorReading, String]
        #OnTimerContext, out: Collector[String]): Unit = { 
        out.collect("温度值连续" + interval/1000 + "秒上升")
        curTimerTsState.clear()
    
      }
    }

测试(前两条和后两条分开输的)  
![在这里插入图片描述][20201021201328475.png_pic_center]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]

[20201021201328475.png_pic_center]: /images/20221123/3fd02e49ec964929b66da45f52781557.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/138e042aff21460a8e2e733b8afe4060.png