Flink的Keyed State支持以下数据类型：

  • ValueState[T] 保存单个的值，值的类型为T。
  • ListState[T] 保存一个列表，列表里的元素的数据类型为T
  • MapState[K, V] 保存Key-Value对
  • ReducingState[T]
  • AggregatingState[I, O]

使用案例: 连续 5s 水位上涨，则告警

声明有状态类型的变量， 这是初始化有 2种方法:

  1. 在 Open 方法 中完成变量的初始化
  2. 在变量声明时使用 lazy val 


  1. package com.ylb.time
  3. import com.ylb.myCluss.WaterSensor
  4. import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
  5. import org.apache.flink.streaming.api.TimeCharacteristic
  6. import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
  7. import org.apache.flink.api.scala._
  8. import org.apache.flink.configuration.Configuration
  9. import org.apache.flink.streaming.api.functions.{AssignerWithPunctuatedWatermarks, KeyedProcessFunction}
  10. import org.apache.flink.streaming.api.watermark.Watermark
  11. import org.apache.flink.util.Collector
  13. /**
  14.   * @author yanglibin
  15.   * @create 2020-03-07 8:56
  16.   */
  17. object watermark_5 {
  18.   def main(args: Array[String]): Unit = {
  19.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  21.     env.setParallelism(1)
  22.     // 设置时间语义
  23.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  25.     // 处理数据
  26.     // 连续 5s 水位上涨，则告警
  28.     // 初始化变量
  29.     // 初始化 当前水位值
  30. //    var currentHigh = 0L
  31.     // 初始化 连续告警值
  32. //    var alarmTimer = 0L
  34.     // markDS
  35.     val markDS: DataStream[WaterSensor] = env.socketTextStream("hadoop-master", 9999)
  36.       // 将数据封装为 WaterSensor 类
  37.       .map(
  38.         line => {
  39.           val datas = line.split(",")
  40.           WaterSensor(datas(0), datas(1).toLong, datas(2).toDouble)
  41.         }
  42.       )
  43.       // 设置 watermark
  44.       .assignTimestampsAndWatermarks(
  45.         new AssignerWithPunctuatedWatermarks[WaterSensor] {
  46.           override def checkAndGetNextWatermark(lastElement: WaterSensor, extractedTimestamp: Long): Watermark = {
  47.             new Watermark(extractedTimestamp)
  48.           }
  50.           override def extractTimestamp(element: WaterSensor, previousElementTimestamp: Long): Long = {
  51.             element.ts * 1000
  52.           }
  53.         }
  54.       )
  56.     // process
  57.     val processDS: DataStream[String] = markDS
  58.       .keyBy(_.id)
  59.       .process(
  60.         new KeyedProcessFunction[String, WaterSensor, String] {
  61.           // 方法1: 在 Open 方法 中完成变量的初始化
  62. //          private var currentHigh:ValueState[Long] = _
  63. //          private var alarmTimer:ValueState[Long] = _
  64. //
  65. //          override def open(parameters: Configuration): Unit = {
  66. //            currentHigh = getRuntimeContext.getState(
  67. //              new ValueStateDescriptor[Long]("currentHigh",classOf[Long])
  68. //            )
  69. //            alarmTimer = getRuntimeContext.getState(
  70. //              new ValueStateDescriptor[Long]("alarmTimer",classOf[Long])
  71. //            )
  72. //          }
  73.           // 方法2: 在变量声明时使用 lazy val
  74.           lazy val currentHigh:ValueState[Long] = getRuntimeContext.getState(
  75.             new ValueStateDescriptor[Long]("currentHigh",classOf[Long])
  76.           )
  77.           lazy val alarmTimer:ValueState[Long] = getRuntimeContext.getState(
  78.             new ValueStateDescriptor[Long]("alarmTimer", classOf[Long])
  79.           )
  81.           override def onTimer(
  82.                                 timestamp: Long,
  83.                                 ctx: KeyedProcessFunction[String, WaterSensor, String]#OnTimerContext,
  84.                                 out: Collector[String]): Unit = {
  85.             out.collect(s"id:${ctx.getCurrentKey}, 当前水位值: ${ctx.timerService().currentWatermark()}, 连接5s水位上涨")
  86.           }
  88.           override def processElement(
  89.                                        value: WaterSensor,
  90.                                        ctx: KeyedProcessFunction[String, WaterSensor, String]#Context,
  91.                                        out: Collector[String]): Unit = {
  92.             if (value.vc > currentHigh.value()) {
  93.               // 如果传入值 大于 当前值, 则注册
  94.               if (alarmTimer.value() == 0) {
  95.                 alarmTimer.update(value.ts * 1000 + 5000)
  96.                 ctx.timerService().registerEventTimeTimer(alarmTimer.value())
  97.               }
  98.             } else {
  99.               // 否则删除 重新注册
  100.               ctx.timerService().deleteEventTimeTimer(alarmTimer.value())
  101.               alarmTimer.update(value.ts * 1000 + 5000)
  102.               ctx.timerService().registerEventTimeTimer(alarmTimer.value())
  103.             }
  105.             currentHigh.update(value.vc.toLong)
  106.           }
  107.         }
  108.       )
  110.     // print
  111.     markDS.print("mark>>")
  112.     processDS.print("process>>")
  115.     // 执行
  116.     env.execute()
  117.   }
  118. }

状态后端

  • MemoryStateBackend
  • FsStateBackend
  • RocksDBStateBackend
    1.       <dependency>
    2.           <groupId>org.apache.flink</groupId>
    3.           <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
    4.           <version>1.7.2</version>
    5.       </dependency>
  1.     val backend:StateBackend = new RocksDBStateBackend("c://tmp/output/flink")
  2.     // 设置状态后端
  3.     env.setStateBackend(backend)
  4.     // 启用检查点
  5.         // 1000ms 生成一条checkpoint数据 + 精准一次性处理
  6.     env.enableCheckpointing(1000,CheckpointingMode.EXACTLY_ONCE)