Flink学习（十六） ProcessFunctionAPI（底层API）

package com.wyh.processFunctionApi

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector


//温度传感器读数样例类
case class SensorReading(id: String, timestamp: Long, temperature: Double)

object ProcessFunctionTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val stream = env.socketTextStream("localhost", 7777)

    //Transform操作
    val dataStream: DataStream[SensorReading] = stream.map(data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })
      //===到来的数据是升序的，准时发车，用assignAscendingTimestamps
      //指定哪个字段是时间戳 需要的是毫秒 * 1000
      //      .assignAscendingTimestamps(_.timestamp * 1000)
      //===处理乱序数据
      //      .assignTimestampsAndWatermarks(new MyAssignerPeriodic())
      //==底层也是周期性生成的一个方法 处理乱序数据 延迟1秒种生成水位 同时分配水位和时间戳 括号里传的是等待延迟的时间
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
      override def extractTimestamp(t: SensorReading): Long = {
        t.timestamp * 1000
      }
    })

    val processedStream = dataStream.keyBy(_.id)
      .process(new TempIncreAlert())

    processedStream.print("process data")


    dataStream.print("input data")


    env.execute("window Test")


  }

}

class TempIncreAlert() extends KeyedProcessFunction[String, SensorReading, String] {
  //定义一个状态,用来保存上一个数据的温度值
  //用懒加载的方式，一开始定义的时候我们还不执行，等到调用的时候去执行
  //所有的状态都这么定义 当成一个变量直接用
  lazy val lastTemp: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))
  //定义一个状态用来保存定时器的时间戳
  lazy val currentTimer: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("currentTimer", classOf[Long]))

  //判断温度连续上升
  //跟上一次数据进行比较 如果比较一直大 10秒种内进行报警
  //注册一个定时器 把上一次的数据保存成当前的状态
  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {
    //每来一条数据的时候，从状态中取出上一次的温度值
    val preTemp = lastTemp.value()
    var curTimerTs = currentTimer.value()

    //更新温度值
    lastTemp.update(value.temperature)

    //加个if判断最开始的温度是否为0来判断是否是第一条数据 温度上升且没有设置过定时器，则注册定时器
    if(preTemp==0.0){
        println("这是第一条数据进来")
    }else if ((value.temperature>preTemp) && (curTimerTs==0L)) {
      
      val timerTs = ctx.timerService().currentProcessingTime() + 10000L

      //传入当前时间加1  是时间戳
      ctx.timerService().registerProcessingTimeTimer(timerTs)
      currentTimer.update(timerTs)
    } else if (value.temperature <= preTemp) {
      //如果温度下降 或者是第一条数据 删除定时器
      ctx.timerService().deleteProcessingTimeTimer(curTimerTs)
      //删除定时器之后将状态清空
      currentTimer.clear()
    }
  }

  //在回调函数中执行定时器到的逻辑
  //当前的时间 ctx上下文 out输出信息
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    //直接输出报警信息
    out.collect(ctx.getCurrentKey + "温度连续上升")
    //考虑真实情况，将状态都清空
    currentTimer.clear()
  }
}

sensor_1,1547718199,39