flink需求之—ProcessFunction（需求：如果30秒内温度连续上升就报警）

1.ProcessFunction是flink中的大杀器（个人认为）

Process Function 用来构建事件驱动的应用以及实现自定义的业务逻辑(使用之前的
window 函数和转换算子无法实现)
Flink 提供了 8 个 Process Function：
 ProcessFunction
 KeyedProcessFunction
 CoProcessFunction
 ProcessJoinFunction
 BroadcastProcessFunction
 KeyedBroadcastProcessFunction
 ProcessWindowFunction
 ProcessAllWindowFunction

2.主要使用KeyedProcessFunction来完成该需求

1.首先介绍一下KeyedProcessFunction

1.KeyedProcessFunction 用来操作 KeyedStream（经过了keyby操作之后）。
KeyedProcessFunction 会处理流的每一个元素，输出为 0 个、1 个或者多个元素。所有的 Process Function 都继承自RichFunction 接口，所以都有 open()、close()和 getRuntimeContext()等方法。而KeyedProcessFunction[KEY, IN, OUT]还额外提供了两个方法:
（1）processElement(v: IN, ctx: Context, out: Collector[OUT]), 流中的每一个元素都会调用这个方法，调用结果将会放在 Collector 数据类型中输出。Context可以访问元素的时间戳，元素的 key，以及 TimerService 时间服务。Context还可以将结果输出到别的流(side outputs)。
（2）onTimer(timestamp: Long, ctx: OnTimerContext, out: Collector[OUT])是一个回调函数。当之前注册的定时器触发时调用。参数 timestamp 为定时器所设定的触发的时间戳。Collector 为输出结果的集合。OnTimerContext 和processElement 的 Context 参数一样，提供了上下文的一些信息，例如定时器触发的时间信息(事件时间或者处理时间)。

2.TimerService 和 定时器（Timers）
（1）Context 和 OnTimerContext 所持有的 TimerService 对象拥有以下方法:
（2）currentProcessingTime(): Long 返回当前处理时间
（3）currentWatermark(): Long 返回当前 watermark 的时间戳
（4）registerProcessingTimeTimer(timestamp: Long): Unit 会注册当前 key 的processing time 的定时器。当 processing time 到达定时时间时，触发 timer。
（5）registerEventTimeTimer(timestamp: Long): Unit 会注册当前 key 的 event time 定时器。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。
（6）deleteProcessingTimeTimer(timestamp: Long): Unit 删除之前注册处理时间定时器。如果没有这个时间戳的定时器，则不执行。
（7） deleteEventTimeTimer(timestamp: Long): Unit 删除之前注册的事件时间定时
器，如果没有此时间戳的定时器，则不执行。

3.当定时器 timer 触发时，会执行回调函数 onTimer()。注意定时器 timer 只能在
keyed streams 上面使用。

2.需求的实现

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector

import java.util.Properties

object ProcessFunctionTest {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "hadoop101:9092")
    val data = env.addSource(new FlinkKafkaConsumer[String]("sensor",new SimpleStringSchema(),properties))

    val dataStream = data.map(x => {
      val arr = x.split(",")
      SensorReading(arr(0).trim, arr(1).trim.toLong, arr(2).trim.toDouble)
    })

    //连续30秒内温度连续上升就报警
    val warningStream = dataStream
      .keyBy(_.id)
      .process(new TempIncreWarning(30000L) )

    warningStream.print("warningStream")
    env.execute()

  }
}

class TempIncreWarning(interval:Long) extends KeyedProcessFunction[String,SensorReading,String] {
  //定义状态，保存上一个温度值进行比较,保存注册定时器的时戳用于删除
  lazy val lastTempState : ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("valueState",classOf[Double]))
  lazy val timerTsState : ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timeState",classOf[Long]))

  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {

    //先取出状态
    val lastTemp = lastTempState.value()
    val timerTs = timerTsState.value()

    lastTempState.update(value.temperature)

    //当前温度与上次温度进行对比
    if (value.temperature > lastTemp && timerTs == 0){
      //如果温度上升并且没有定时器，那么注册当前数据时间戳10s之后的定时器
      val ts = ctx.timerService().currentProcessingTime() + interval
      ctx.timerService().registerProcessingTimeTimer(ts)
      timerTsState.update(ts)

    }else if(value.temperature < lastTemp){
      ctx.timerService().deleteProcessingTimeTimer(timerTs)
      timerTsState.clear()
    }

  }
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    out.collect("传感器" + ctx.getCurrentKey + "的温度连续" + interval/1000 + "秒连续上升" )
    timerTsState.clear()
  }

}

3.结果展示

输入条件：
>"sensor_1", 1547718199, 35.5
>"sensor_1", 1547718199, 35.6
>"sensor_1", 1547718199, 35.7
>"sensor_1", 1547718199, 35.8
>"sensor_1", 1547718199, 35.9
>"sensor_1", 1547718199, 36.1
>"sensor_1", 1547718199, 36.2
>"sensor_1", 1547718199, 36.3
>"sensor_1", 1547718199, 36.4
>"sensor_1", 1547718199, 36.5
>

输出结果：