文章目录
- 知识点
- 反压
- Checkpoint
- Barrier
- Aligned Checkpoint
- Unaligned Checkpoint
- 核心思想
- 实现原理
- UC同步阶段
- UC异步阶段
知识点
反压
反压是流式系统中关于处理能力的动态反馈机制,并且是从下游到上游的反馈,一般是在实时数据处理的过程中,上游节点的生产速度大于下游节点的消费速度。在Flink中,反压主要有两个部分:跨TaskManager的反压过程和TaskManager内的反压过程。
https://blog.csdn.net/Stray_Lambs/article/details/120578294
Checkpoint
Flink 借助Checkpoint机制来保证有状态的分布式计算
所谓Checkpoint,就是周期性的进行Snapshot的过程
当算子Failover的时候,通过快照恢复算子的状态,
Flink的Checkpoint是基于Chandy-Lamport(CL)算法改进的
ps:
Spark中Shuffle是划分Stage的边界,每个Shuffle阶段数据会进行分区、排序、聚合、写入磁盘等操作,自然的进行了状态的保存。所以Spark中无复杂的Checkpoint机制。
Barrier
先说明一下barrier这个概念
以一定的间隔被插入到Source节点,经过若干个算子,不断的向下游传递。
当barrier到达算子的时候,触发Snapshot,当前算子的状态保存完毕后,传递到下一个算子。
当一个算子对应多个输入时,也就意味着这个算子会接收到多个barrier,此时为了保证全局一致性,那么就需要对齐多个输入的barrier,这个就是Aligned Checkpoint
Aligned Checkpoint
存在的问题:
- 对齐时间长,反压时被完全阻塞。
原因:每条链路的处理速度是不一样的,barrier对齐需要时间,如果某一条链路有反压,会因为需要对齐这个过程,使得整个链路被阻塞,因为barrier没有对齐,为了保障数据一致性,checkpoint也无法进行,接下来Checkpoint超时失败,重新进行Checkpoint,但是由于反压仍然存在,最终陷入失败–重启–失败的循环
Unaligned Checkpoint
为了解决Aligned Checkpoint中存在的反压严重时Checkpoint失败的问题
提出了Unaligned Checkpoint
核心思想
允许Barrier超越ongoing data(正在进行的数据,在buffer中,还没有进入task的数据,比如下图中蓝色的2、3、4、5),barrier超越了这些数据,那如果算子Failover的话,这些数据不就丢失了,解决的方法是,将这些数据也进行快照,在Failover时重放这部分数据。
barrier1顺利超越ongoing data,进入到task中,算子收到全部task后开始进行snapshot
实现原理
假设当前task的上游并行度为3,下游并行度为2,如上图所示,task有3个输入和两个输出,矩形表示buffer中的一条一条数据
Unaligned Checkpoint 这里简称UC
整个UC分为UC同步阶段、UC异步阶段两部分
UC同步阶段
UC开始后,task的3个input-buffer会陆续收到上游发送的barrier,如图所示input-buffer1收到了barrier1,其它的input-buffer还没有收到barrier,当某一个input-buffer接收到barrier时,task会直接开始UC的第一阶段,即UC同步阶段。
这个阶段只需要三个input-buffer中任意一个buffer的barrier进入到task的网络缓冲内存中,task就会直接开始UC,不用等其它的Input-buffer接收到barrier,也不需要处理完input-buffer1接收到barrier1之前的数据。
Flink 网络缓冲区
https://www.jianshu.com/p/cfbb0cf69ae3
UC同步阶段:barrier超越ongoing data
如下图,可以看到barrier超越了input-buffer和output-buffer中的所有数据,到达下游output-buffer的头部,被快速的传递给下游的task,这也解释了为什么在反压情况下UC可以成功
从task层面来看,barrier可以在task内部实现快速超车
从Job层面来看,如果每一个task内部,barrier都可以快速超车,那么barrier就可以从source task快速的超车到Sink task
为了保证数据一致性,UC同步阶段,task不能处理数据
UC同步阶段的四个主要流程
- barrier超车,当算子的某个input-buffer接收到barrier时,超越ongoing data,快速的将其传递到output-buffer的头部,保证其可以快速到达下游算子
- buffer引用,对buffer进行引用,这里不进行snapshot,真正的快照在UC异步阶段进行
- 调用task的SnapshotState方法
- StateBackend同步快照
UC异步阶段
UC同步阶段完成后,task继续处理数据,同时进行UC的第二个阶段,barrier对齐和UC异步阶段。
首先异步阶段要快照同步阶段所有引用的input-buffer和output-buffer以及同步阶段算子内部引用的State
UC异步阶段其实也有barrier对齐,当task开始UC的时候,很多input-buffer没有接收到barrier,这些input-buffer之前可能还有一些buffer需要快照,例如上图的绿色数据块,所以UC异步阶段要等到所有的input-buffer barrier都到达,且barrier之前的所有buffer都需要快照,这就是UC异步阶段的barrier对齐
这个对齐过程理论上会很快,因为链路中的每一个task,barrier都可以快速的超越所有input-buffer、output-buffer,优先传递barrier到下游task。
异步阶段需要写三部分数据到DFS,分别是UC同步阶段引用的算子内部的State、同步阶段引用的所有input-buffer和output-buffer、以及其它input-buffer barrier之前的buffer
这三部分数据写完之后,task会将结果汇报到TaskManager。
资料:
Flink Unaligned Checkpoint 在 Shopee 的优化和实践