Flink状态管理

Posted 2020-11-26 shengyang17

 

 

Flink中的状态

• 算子状态（Operatior State）
• 键控状态（Keyed State）
• 状态后端（State Backends）

有状态的流式处理（可以是有状态（如一段时间内温度连续上升就报警；wordcount计算）、也可以是无转态（基于某个独立数据事件做处理直接输出，来一个处理一个，如map、flatmap、filter））

由一个任务维护，并且用来计算某个结果的所有数据，都属于这个任务的转态；

可以认为状态就是一个本地变量，可以被任务的业务逻辑访问；

Flink会进行状态管理（状态做序列化以二进制的形式全部存储起来），包括状态一致性、故障处理以及高效存储和访问，以便开发人员可以专注于应用程序的逻辑。

 

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

有两种状态：

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

　　键控状态（Keyed State），根据输入数据流中定义的键（Key）来维护和访问（不同的key也是独立访问的）；

算子状态：

 

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

状态对于同一个任务而言是共享的；

算子状态不能由相同或不同算子的另一个任务访问；

 

算子状态数据结构：

 

列表状态（List state），将状态表示为一组数据的列表；

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

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

 

 

 

 

 

 键控状态（Keyed State）

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

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

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

 

 键控状态数据结构：

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

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

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

聚合状态（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)

 

状态后端（State Backends）

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

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

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

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

 

选择一个状态后端

MemoryStateBackend：

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

FsStateBackend（文件系统状态后端）：

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

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

RocksDBStateBackend：

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