Flink| 状态管理| 状态编程

1. 状态管理

Flink中的状态

A. 算子状态（Operatior State）
B. 键控状态（Keyed State）
状态后端（State Backends） -- 用作状态管理

流式处理（A. 可以是无状态（基于某个独立的事件计算出来后直接输出了，来一个处理一个不涉及到其他东西，如map、flatmap、filter；超过一定温度就报警 - 侧输出流；）、

B. 可以是有转态的（求和、wordcount计算））

状态是针对一个任务而言的，由一个任务维护，并且用来计算某个结果的所有数据，都属于这个任务的转态；
可以认为状态就是一个本地变量，可以被任务的业务逻辑直接访问；
Flink会进行状态管理（状态做序列化以二进制的形式全部存储起来），包括状态一致性、故障处理以及高效存储和访问，以便开发人员可以专注于应用程序的逻辑。

在Flink中，状态始终与特定算子相关联；为了运行时的Flink了解算子的状态，算子需要预先注册其状态；

有两种状态：

　　算子状态（Operator State），算子状态的作用范围限定为算子任务，一个任务一个状态；

　　键控（分区）状态（Keyed State），根据输入数据流中定义的键（Key）来维护和访问（基于KeyBy--KeyedStream上有任务出现的状态，定义的不同的key来维护这个状态；不同的key也是独立访问的，一个key只能访问它自己的状态，不同key之间也不能互相访问）；

A. 算子状态：

算子状态的作用范围限定为算子任务，由同一并行子任务所处理的所有数据都可以访问到相同的状态；

状态对于同一个任务而言是共享的（每一个并行的子任务共享一个状态）；

算子状态不能由相同或不同算子的另一个任务访问（相同算子的不同任务之间也不能访问）；

算子状态数据结构：

列表状态（List state），将状态表示为一组数据的列表；（会根据并行度的调整把之前的状态重新分组重新分配）

联合列表状态（Union list state），也将状态表示为数据的列表，它常规列表状态的区别在于，在发生故障时，或者从保存点（savepoint）启动应用程序时如何恢复（把之前的每一个状态广播到对应的每个算子中）。

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

B. 键控状态（Keyed State）-- 更常用

键控状态是根据输入数据流中定义的键（key）来维度和访问状态的；

Flink为每个key维护一个状态实例，并将具有相同键的所有数据，都分区到同一个算子任务中，这个任务会维护和处理这个key对应的状态；

当任务处理一条数据时，它会自动将状态的访问范围限定为当前数据的key；

键控状态数据结构：

值状态（Value state），将状态表示为单个值；（直接.value获取，Set操作是.update）

列表状态（List state），将状态表示为一组数据的列表（存多个状态）；（.get，.update，.add）

映射状态（Map state），将状态表示为一组Key-Value对；（.get，.put ，类似HashMap）

聚合状态（Reducing state & Aggregating State），将状态表示为一个用于聚合操作的列表；（.add不像之前添加到列表，它是直接聚合到之前的结果中）

　　　　Reduce输入输出类型是不能变的，Aggregate可得到数据类型完全不一样的结果；

键控状态的使用：

声明一个键控状态:
lazy val lastTemp: ValueState[Double] = getRuntimeContext.getState(
  new ValueStateDescriptor[Double]("lastTemp", classOf[Double])
)
读取状态:
val prevTemp = lastTemp.value()
对状态赋值:
lastTemp.update(value.temperature)

C. 状态后端（State Backends） -- 状态管理（存储、访问、维护和检查点）

每传入一条数据，有状态的算子任务都会读取和更新状态；

由于有效的状态访问对于处理数据的低效迟至关重要，因此每个并行任务都会在本地维度其状态，以确保快速的状态访问；

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

状态后端主要负责两件事：本地的状态管理，以及将检查点（checkpoint）状态写入远程存储；

状态后端的分类：

MemoryStateBackend：一般用于开发和测试

内存级的状态后端，会将键控状态作为内存中的对象进行管理，将它们存储在TaskManager的JVM堆上，而将checkpoint存储在JobManager的内存中；
特点快速、低延迟，但不稳定；

FsStateBackend（文件系统状态后端）：生产环境

将checkpoint存到远程的持久化文件系统（FileSystem）上，而对于本地状态，跟MemoryStateBackend一样，也会存到TaskManager的JVM堆上。

同时拥有内存级的本地访问速度，和更好的容错保证；（如果是超大规模的需要保存还是无法解决，存到本地状态就可能OOM）

RocksDBStateBackend：

将所有状态序列化后，存入本地的RocksDB（本地数据库硬盘空间）中存储，全部序列化存储到本地。

2. 状态编程

package com.xxx.fink.api.windowapi

import akka.pattern.BackoffSupervisor.RestartCount
import com.xxx.fink.api.sourceapi.SensorReading
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.runtime.executiongraph.restart.RestartStrategy
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}


/**
  *状态编程
  * 检测两次温度变化如果超过某个范围就报警，比如超过10°就报警；
  */
object StateTest {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    //env.setStateBackend(new RocksDBStateBackend(""))
    val stream: DataStream[String] = env.socketTextStream("hadoop101", 7777)

    val dataStream: DataStream[SensorReading] = stream.map(data => {
      val dataArray: Array[String] = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000
      })

    //方法一:
    val processedStream1: DataStream[(String, Double, Double)] = dataStream.keyBy(_.id)
        .process(new TempChangeAlert(10.0))

    //方法二: 除了processFunction，其他也可以有状态
    val processedStream2: DataStream[(String, Double, Double)] = dataStream.keyBy(_.id)
      .flatMap(new TempChangeAlert2(10.0))

    //方法三: 带状态的flatMap
    val processedStream3: DataStream[(String, Double, Double)] = dataStream.keyBy(_.id)
        .flatMapWithState[(String, Double, Double), Double]{
      //如果没有状态的话，也就是没有数据过来，那么就将当前数据湿度值存入状态
      case (input: SensorReading, None) => (List.empty, Some(input.temperature))
      //如果有状态，就应该与上次的温度值比较差值，如果大于阈值就输出报警
      case(input: SensorReading, lastTemp: Some[Double]) =>
        val diff = (input.temperature - lastTemp.get).abs
        if (diff > 10.0){
          (List((input.id, lastTemp.get, input.temperature)), Some(input.temperature))
        }else{
          (List.empty, Some(input.temperature))
        }
    }

    dataStream.print("Input data:")
    processedStream3.print("process data:")

    env.execute("Window test")

  }

}

class TempChangeAlert(threshold: Double) extends KeyedProcessFunction[String, SensorReading, (String, Double, Double)] {
  //定义一个状态变量，保存上次的温度值
  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))

  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, (String, Double, Double)]#Context, out: Collector[(String, Double, Double)]): Unit = {
    //获取上次的温度值
    val lastTemp = lastTempState.value()
    //用当前的温度值和上次的做差，如果大于阈值，则输出报警
    val diff = (value.temperature - lastTemp).abs
    if (diff > threshold){
      out.collect(value.id, lastTemp, value.temperature)
    }
    lastTempState.update(value.temperature) //状态更新

  }
}

class TempChangeAlert2(threshold: Double) extends RichFlatMapFunction[SensorReading, (String, Double, Double)]{
  //flatMap本身是无状态的，富函数版本的函数类都可以去操作状态也有生命周期
  private var lastTempState: ValueState[Double] = _ //赋一个空值；
  //初始化的声明state变量
  override def open(parameters: Configuration): Unit = {
  //可以定义一个lazy；也可以在声明周期中拿；
    lastTempState = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))
  }

  override def flatMap(value: SensorReading, out: Collector[(String, Double, Double)]): Unit = {
    //获取上次的温度值
    val lastTemp = lastTempState.value()
    //用当前温度值和上次的求差值，如果大于阈值，输出报警信息
    val diff = (value.temperature - lastTemp).abs
    if (diff > threshold){
      out.collect(value.id, lastTemp, value.temperature)
    }
    lastTempState.update(value.temperature)
  }
}

测试：

###方法二测试：
Input data:> SensorReading(sensor_1,1547718199,35.8)
process data:> (sensor_1,0.0,35.8)
Input data:> SensorReading(sensor_1,1547718199,32.0)
Input data:> SensorReading(sensor_1,1547718199,25.0)
Input data:> SensorReading(sensor_1,1547718199,35.1)
process data:> (sensor_1,25.0,35.1)
Input data:> SensorReading(sensor_1,1547718199,12.0)
process data:> (sensor_1,35.1,12.0)

###方法三：
Input data:> SensorReading(sensor_1,1547718199,35.8)
Input data:> SensorReading(sensor_1,1547718199,25.0)
process data:> (sensor_1,35.8,25.0)
Input data:> SensorReading(sensor_1,1547718199,28.8)
Input data:> SensorReading(sensor_1,1547718199,39.8)
process data:> (sensor_1,28.8,39.8)