Flink原理学习状态

Posted 2023-02-23 oahaijgnahz

Flink原理学习之 状态

---

文章目录

• Flink原理学习之 状态
• 一、Flink的状态
• 二、Flink自身容错——checkpoint机制
• 三、Flink端到端一致性保证
• 四、Flink应用算子并行度发生变化后State恢复

---

Java、大数据开发学习要点（持续更新中…）

---

一、Flink的状态

  状态在Flink中叫做State，用来保存中间计算结果或者缓存数据。State是实现有状态的计算下Exactly-Once的基础。

1. 状态主要分为两种类型：

• Operator State：跟一个特定算子的实例绑定，整个算子只对应一个State对象(相同的并行算子都能访问到状态)。只支持ListState
• Keyed State：状态跟特定的key绑定，即每个KeyedStream对应一个State对象。支持多种State(ValueState、MapState、ListState等)

2. 广播状态：
   在一个数据流需要被广播到所有下游任务，并在下游算子本地存储，在处理另一个流的时候就要依赖广播状态。比如业务数据流正常处理，但当规则更新的时候，则规则数据流需要被广播至所有业务数据流的并行算子中，并以广播State的形式在这些算子中缓存规则数据(必须是MapState类型)。主流和广播流通过connect，广播函数也提供了处理广播流数据和主流数据的接口。

3. 状态后端：

   状态后端具备以下两种能力：
   (1) 在计算过程中提供访问State的能力，即在程序中能够通过StateBackend的接口读取数据。
   (2)能够将State持久化到外部存储中，提供容错能力(HDFS、RocksDB等)

   状态的持久化分为 全量持久化 和 增量持久化(只有RocksDB支持)：
   全量持久化通过将状态CopyOnWrite来保证并发安全，而RocksDB通过自身提供的快照机制来实现。RocksDB实际是基于LSM树的KV存储数据库，也就是说内存和磁盘数据都是有序的，因此可以实现增量存储。

4. 状态描述：
   状态描述为状态指定名称、状态中存储的类型、TTL等信息。用于保存在StateBackend中，对状态进行区分的依据。这样可以在ProcessFunction或者RichFunction类中通过RuntimeContext上下文从StateBackend中获取实际的State实例。

二、Flink自身容错——checkpoint机制

  Flink采用轻量级分布式快照实现应用容错。Flink的分布式快照实现是通过向数据流中注入隶属于不同快照的Barrier(屏障)来切分数据流，从而实现并发异步分布式快照。Barrier周期性注入数据流中，作为数据流的一部分，从上游至下游被算子处理。
  Barrier(携带ID)会在数据流Source中被注入并行数据流中。对于Kafka作为数据源的情况，Barrier注入位置就是Source消费Kafka对应分区的数据的最后一个偏移量，Source算子在读取到对应Barrier后会携带当前读到的偏移量做checkpoint，由此，任务恢复时也是从保存的checkpoint中记录的偏移量再继续从Kafka读取数据。
  引擎内Exactly-Once的实现：当Sink算子收到了上游算子的某个Barrier，Sink算子对自己的状态进行checkpoint，然后通知JobMaster(其中的CheckpointCoordinator)，当所有并行算子都向JobMaster汇报成功后，JobMaster向所有算子确认本次快照完成。当其中某个Task出现故障，那么Flink的此作业暂停，并将各个算子的状态恢复到上一个检查点的状态，并重新从检查点保存的Kafka偏移量开始消费数据。（可以看到，虽然Kafka Consumer是没有实现事务的，但Flink可以通过自己的checkpoint机制来实现消费Kafka的事务保证）

当某个算子上游有多个数据流时，需要进行barrier对齐，做法是barrier先到达的数据流之后的数据缓存在内存中不处理，等到其他barrier到达后，将barrier向后传递并进行异步快照。

三、Flink端到端一致性保证

  由于Flink内部由checkpoint机制保证了数据的一致性，Source状态的持久化能够记录读取到数据源的位置，当然这需要数据源可重设数据读取位置(Kafka、Flume的taildir Source)。而Flink的Sink需要保证事务避免数据重复写入，能够支持Flink的端到端一致性要求Sink满足支持回滚机制或者满足幂等写入(Kafka、Redis、HBase)。
  真正实现Sink写入的Exactly-Once是通过Sink的两阶段提交(2PC)实现的。如前面checkpoint流程所说，当Barrier到达Sink后，Sink会做checkpoint并开启下游设备的一个事务将接下来的数据写入该事务中(预提交，根据事务的隔离性当前的数据对存储系统不可见)。然后，JobMaster收到Sink的checkpoint完成后向所有并行算子发送确认检查点完成的通知，如果收到所有算子都完成则正式提交。如果正式提交过程中，如果因为网络原因commit请求失败，那么Flink会回滚到预提交的那个checkpoint后重新进行commit请求。

---

实际当Kafka是数据输出的Sink，Flink保证了数据的端到端一致性，即使Flink不能保证，Kafka如果开启事务也能保证写入数据的Exactly-Once

四、Flink应用算子并行度发生变化后State恢复

State分为Operator State和Keyed State，在并行度改变的时候，会遵循不同的规则，调整State，在调整的过程中，尽量保证State数据是在本地的，但是某些情况下会导致State数据在全局环境中，进行重新Shuffle导致性能损耗。

由于checkpoint是分散在每个task上执行，再做汇总持久化。这些task做的checkpoint数据在后面应用恢复时包括并行度扩增或减少时能够重新打散分布。