我们都爱学的flink的state内容

一、概述

我们来看一下flink的介绍，

Apache Flink® — Stateful Computations over Data Streams，数据流上的状态计算。可以看出flink默认它是一个默认就有状态的分析引擎，State一般指一个具体的 Task/Operator 的状态，State数据默认保存在 Java 的堆内存中。

假设一个 Task 在处理过程中挂掉了，那么它在内存中的状态都会丢失，所有的数据都需要重新计算。从容错和消息处理的语义（At -least-once 和 Exactly-once）上来说，Flink引入了State 和 CheckPoint。

例如：WordCount 案例可以做到单词的数量的累加，其实是因为在内存中保证了每个单词的出现的次数，这些数据其实就是状态数据。

现在你大概知道了state是什么，那么这些 state 存储的介质有哪些？ Flink 提供了三种存储 State的介质。

使用方法：

MemoryStateBackend( int maxStateSize, boolean asynchronousSnapshots )

存储位置：

State: TaskManager 内存

Checkpoint： Jobmanager 内存

使用场景：

本地测试用，不推荐生产场景使用

使用方法：

FaStateBackend( URI checkpointDataUri, boolean asynchronousSnapshots )

存储位置：

State：Taskmanager 内存

Checkpoint: 外部文件系统( 本地或 HDFS )

使用场景：

常规使用 State 的作业，可以在生产中使用

使用方法：

RocksDBStateBackend( URI checkpointDataUri, boolean enableIncrementalCheckpointing )

存储位置：

State: TaskManager 上的 KV 数据库（实际使用内存 + 磁盘）

Checkpoint: 外部文件系统（本地或 HDFS )

使用场景：

超大状态作业，对性能要求不高的生产场景

二、state的类型

Flink中有原生状态、托管状态两种基本类型的State，这两种类型的state都可以使用Operator State(算子状态)、keyed State(键控状态)这两种形式操作。

由用户操作算子自己管理数据结构，当触发Checkpoint操作过程中，Flink并不知道状态数据内部的数据结构，只是将数据转换成bytes数据存储在Checkpoints中，当从Checkpoints恢复任务时，算子自己再反序列化出状态的数据结构。

由Flink Runtime控制和管理状态数据，并将状态数据转换成为内存的Hash tables或 RocksDB的对象存储，然后将这些数据通过内部的接口持久化到checkpoints中，如果任务发生异常时，可以通过这些状态数据恢复任务。

下面是两种状态的比较表格，推荐使用ManagedState管理状态数据，ManagedState更好的支持状态数据的重平衡以及更加完善的内存管理

operator state是task级别的state，说白了就是每个task对应一个state。

Kafka Connector source中的每个分区（task）都需要记录消费的topic的partition和offset等信息。

对于Operator State，我们还需进一步实现CheckpointedFunction接口。

它经常被用在Source或Sink等算子上，用来保存流入数据的偏移量或对输出数据做缓存，以保证Flink应用的Exactly-Once语义。

keyed State使用场景广泛，它是基于KeyedStream上的状态，这个状态是跟特定的Key 绑定的。KeyedStream流上的每一个Key，都对应一个State。Flink针对 Keyed State 提供了下面几种数据结构的（托管状态）Keyed state保存State。

保存一个可以更新和检索的值（例如：每个值都对应到当前的输入数据的key，因此算子接收到的每个key都可能对应一个值）。 这个值可以通过update(T) 进行更新，通过 T value() 进行检索 。

保存一个元素的列表。可以往这个列表中追加数据，并在当前的列表上进行检索。可以通过 add(T) 或者 addAll(List) 进行添加元素，通过 Iterable get() 获得整个列表。还可以通过 update(List) 覆盖当前的列表 。

维护了一个映射列表。 你可以添加键值对到状态中，也可以获得 反映当前所有映射的迭代器。使用 put(UK，UV) 或者 putAll(Map) 添加映射。 使用 get(UK) 检索特定 key。 使用 entries()，keys() 和 values() 分别检索映射、 键和值的可迭代视图。

保存一个单值，表示添加到状态的所有值的聚合。接口与ListState类似，但使用add(T) 增加元素，会使用提供的 ReduceFunction 进行聚合。

AggregatingState: 保留一个单值，表示添加到状态的所有值的聚合。和 ReducingState 相反的是, 聚合类型可能与添加到状态的元素的类型不同。 接口与 ListState类似，但使用 add(IN) 添加的元素会用指定的 AggregateFunction 进行聚合

1、只能用于RichFunction

2、将State 声明为实例变量

3、在 open() 方法中为State赋值创建一个StateDescriptor利用getRuntimeContext().getXXState(…)构建不同的State

4、调用State的方法进行读写例如 state.value()、state.update(…)等等

Flink 大数据 实时流计算服务 CS

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。