Flink实例(三十):状态管理(一)概述

参考链接;

https://www.cnblogs.com/qiu-hua/p/13432337.html

https://www.cnblogs.com/shengyang17/p/12549199.html

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之间也不能互相访问);

2 算子状态:

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

状态对于同一个任务而言是共享的(每一个并行的子任务共享一个状态);

算子状态不能由相同或不同算子的另一个任务访问(相同算子的不同任务之间也不能访问);

算子状态数据结构:

列表状态(List state),将状态表示为一组数据的列表;(会根据并行度的调整把之前的状态重新分组重新分配

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

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

补充:

Operator State是task级别的state,说白了就是每个task对应一个state(要明白flink中的task是指什么哈)
例如Kafka Connector source中的每个分区(task)都需要记录消费的topic的partition和offset等信息。

3 键控状态(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可得到数据类型完全不一样的结果;

StateDescriptor

   State 既然是暴露给用户的,那么就需要有一些属性需要指定:state 名称、val serializer、state type info。

  在对应的statebackend中,会去调用对应的create方法获取到stateDescriptor中的值。

  Flink通过StateDescriptor来定义一个状态。这是一个抽象类,内部定义了状态名称、类型、序列化器等基础信息。与上面的状态对应,从StateDescriptor派生了ValueStateDescriptorListStateDescriptor等descriptor

  • ValueState getState(ValueStateDescriptor)
  • ReducingState getReducingState(ReducingStateDescriptor)
  • ListState getListState(ListStateDescriptor)
  • FoldingState getFoldingState(FoldingStateDescriptor)
  • MapState getMapState(MapStateDescriptor)

最后来一张较为全面的图

键控状态的使用:

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

4 状态后端(State Backends) -- 状态管理(存储、访问、维护和检查点)

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

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

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

状态后端主要负责两件事:本地的状态管理,以及将检查点(checkpoint)状态写入远程存储;

状态后端的分类:

MemoryStateBackend: 一般用于开发和测试

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

FsStateBackend(文件系统状态后端):生产环境

  • 将checkpoint存到远程的持久化文件系统(FileSystem)上,而对于本地状态,跟MemoryStateBackend一样,也会存到TaskManager的JVM堆上。
  • 同时拥有内存级的本地访问速度,和更好的容错保证;(如果是超大规模的需要保存还是无法解决,存到本地状态就可能OOM)

RocksDBStateBackend:

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

5. state总结

1. 是否存在当前处理的 key(current key):operator state 是没有当前 key 的概念,而 keyedstate 的数值总是与一个 current key 对应。
2. 存储对象是否 on heap: 目前 operator state backend 仅有一种 on-heap 的实现;而 keyed statebackend 有 on-heap 和 offff-heap(RocksDB)的多种实现。
3. 是否需要手动声明快照(snapshot)和恢复 (restore) 方法:operator state 需要手动实现snapshot 和 restore 方法;而 keyed state 则由 backend 自行实现,对用户透明。
4. 数据大小:一般而言,我们认为 operator state 的数据规模是比较小的;认为 keyed state 规模是相对比较大的。需要注意的是,这是一个经验判断,不是一个绝对的判断区分标准。

本文来自博客园,作者:秋华,转载请注明原文链接:https://www.cnblogs.com/qiu-hua/p/13794502.html

原文地址:https://www.cnblogs.com/qiu-hua/p/13794502.html