Flink 有状态的算子和应用程序

流式计算分为无状态和有状态两种情况。无状态的计算观察每个独立事件，并根据最后一个事件输出结果。例如，流处理应用程序从传感器接收水位数据，并在水位超过指定高度时发出警告。有状态的计算则会基于多个事件输出结果。以下是一些例子。

• 所有类型的窗口。例如，计算过去一小时的平均水位，就是有状态的计算。
• 所有用于复杂事件处理的状态机。例如，若在一分钟内收到两个相差20cm以上的水位差读数，则发出警告，这是有状态的计算。
• 流与流之间的所有关联操作，以及流与静态表或动态表之间的关联操作，都是有状态的计算。

无状态流处理和有状态流处理的主要区别：

• 无状态处理分别接收每条数据记录（黑条），然后根据最新输入的数据生成输出数据（白条）
• 有状态流处理会维护状态（根据每条输入记录进行更新），并给予最新输入的记录和当前的状态值生成输出记录（灰条）

尽管无状态的计算很重要，但是流处理对有状态的计算更感兴趣。事实上，正确地实现有状态的计算比实现无状态的计算难得多。旧的流处理系统并不支持有状态的计算，而新一代的流处理系统则将状态及其正确性视为重中之重。

1、 算子状态（operator state）

Flink内置的很多算子，数据源source，数据存储sink都是有状态的，流中的数据都是buffer records，会保存一定的元素或者元数据。例如: ProcessWindowFunction会缓存输入流的数据，ProcessFunction会保存设置的定时器信息等等。

在Flink中，状态始终与特定算子相关联。总的来说，有两种类型的状态：

• 算子状态（operator state）
• 键控状态（keyed state）

1.1.1 算子状态 （operator state）

算子状态的作用范围限定为算子任务。这意味着由同一并行任务所处理的所有数据都可以访问到相同的状态，状态对于同一任务而言是共享的。算子状态不能由相同或不同算子的另一个任务访问。

Flink为算子状态提供三种基本数据结构：

（1）列表状态（List state）

将状态表示为一组数据的列表。

（2）联合列表状态（Union list state）

也将状态表示为数据的列表。它与常规列表状态的区别在于，在发生故障时，或者从保存点（savepoint）启动应用程序时如何恢复。

（3）广播状态（Broadcast state）

如果一个算子有多项任务，而它的每项任务状态又都相同，那么这种特殊情况最适合应用广播状态。

1.1.2 键控状态（keyed state）

2、 键控状态 （keyed state)

键控状态是根据输入数据流中定义的键（key）来维护和访问的。Flink为每个键值维护一个状态实例，并将具有相同键的所有数据，都分区到同一个算子任务中，这个任务会维护和处理这个key对应的状态。当任务处理一条数据时，它会自动将状态的访问范围限定为当前数据的key。因此，具有相同key的所有数据都会访问相同的状态。Keyed State很类似于一个分布式的key-value map数据结构，只能用于KeyedStream（keyBy算子处理之后）。

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

• ValueState[T] 保存单个值，值得类型为T
  • get操作: ValueState.value()
  • set操作: ValueState.update(value: T)
• ListState[T]保存一个列表，列表里的元素的数据类型为T
  • ListState.add(value: T)
  • ListState.addAll(values: java.util.List[T])
  • ListState.get()返回Iterable[T]
  • ListState.update(values: java.util.List[T])
• MapState[K, V]保存Key-Value对
  • MapState.get(key: K)
  • MapState.put(key: K, value: V)
  • MapState.contains(key: K)
  • MapState.remove(key: K)
• ReducingState[T]
• AggregatingState[I, O]

State.clear()是清空

通过RuntimeContext注册StateDescriptor。

StateDescriptor以状态state的名字和存储的数据类型为参数

在open方法中创建state变量（如果直接初始化，会抛出异常，Exception in thread "main" java.lang.IllegalStateException: The runtime context has not been initialized.）

需求：如果连续两次水位差超过40cm，发生预警信息。

object AlarmTest {
    def main(args: Array[String]): Unit = {
       
        // TODO 需求:如果连续两次水位差超过40cm，发生预警信息。
        
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment;
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        
        //val dataDS: DataStream[String] = env.readTextFile("input/sensor-data2.log")
        val dataDS: DataStream[String] = env.socketTextStream("linux1", 9999)
        val waterDS = dataDS.map(
            data=>{
                val datas = data.split(",")
                WaterSensor(datas(0), datas(1).toLong, datas(2).toInt)
            }
        )
    
        // 设定数据的事件时间已经定义Watermark
        val markDS: DataStream[WaterSensor] = waterDS.assignAscendingTimestamps(_.ts * 1000)
    
        // TODO 对分区后的数据进行处理
        markDS.keyBy(_.id)
                .process( new KeyedProcessFunction[String, WaterSensor, String] {
    
                    private var lastWaterVal : ValueState[Int] = _
    
                    override def open(parameters: Configuration): Unit = {
                        lastWaterVal = getRuntimeContext.getState[Int](
                            new ValueStateDescriptor[Int]("lastWaterVal", classOf[Int])
                        )
                    }
    
                    // TODO 当水位差超过40cm，马上预警
                    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, WaterSensor, String]#OnTimerContext, out: Collector[String]): Unit = {
                        out.collect("水位差超过40cm")
                    }
    
                    override def processElement(value: WaterSensor, ctx: KeyedProcessFunction[String, WaterSensor, String]#Context, out: Collector[String]): Unit = {
                        // TODO 当前水位应该减去上一次记录的水位是否超过40cm
                        if ( (value.vc - lastWaterVal.value()) > 40 ) {
                            ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 1000)
                        }
                        lastWaterVal.update(value.vc)
                        
                    }
                } ).print("alarm>>>>")
        markDS.print("mark>>>>>>>")
        env.execute()      
    }   
}

也可以用lazy懒加载方式，

 private lazy var lastWaterVal : ValueState[Int] =getRuntimeContext.getState[Int](
                            new ValueStateDescriptor[Int]("lastWaterVal", classOf[Int])

3、 状态后端（state Backend）

每传入一条数据，有状态的算子任务都会读取和更新状态。由于有效的状态访问对于处理数据的低延迟至关重要，因此每个并行任务都会在本地维护其状态，以确保快速的状态访问。

状态的存储、访问以及维护，由一个可插入的组件决定，这个组件就叫做状态后端（state backend）

状态后端主要负责两件事：

• 本地的状态管理
• 将检查点（checkpoint）状态写入远程存储

状态后端分类：

（1） MemoryStateBackend

内存级的状态后端，会将键控状态作为内存中的对象进行管理，将它们存储在TaskManager的JVM堆上；

将checkpoint存储在JobManager的内存中。

（2）FsStateBackend

本地状态，跟MemoryStateBackend一样，也会存在TaskManager的JVM堆上

将checkpoint存到远程的持久化文件系统（FileSystem）上

（3）RocksDBStateBackend

将所有状态序列化后，存入本地的RocksDB中存储

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
    <version>1.7.2</version>
</dependency>

设置状态后端为RocksDBStateBackend：

val checkpointPath : String = "XXX路径"
val stateBackend:StateBackend = new RocksDBStateBackend(checkpointPath)
env.setStateBackend(stateBackend)
env.enableCheckpointing(1000, CheckpointingMode.EXACTLY_ONCE)