![cover](https://img-blog.csdnimg.cn/img_convert/8db9ff7bf6ed39b2a087509709fe315c.png)
什么是Flink的非barrier对齐,如何使用?
拥有两个输入流的 operators(例如 CoProcessFunction)会执行 barrier 对齐(barrier alignment),以便当前快照能够包含两个输入流 barrier 之前(但不超过)的所有 events 产生的状态。字母流的 barrier n 前面的数据(abcd)会被保存到状态里面,直到 barrier n 到来以后,再进行checkpoint,将数据保存到检查点
众所周知,flink在开启checkpoint之后,source 任务收到触发检查点保存的指令后,会立即在当前处理的数据中插入一个标识字段(Barrier),然后再向下游任务发出。我们平时使用比较多的是对齐的barrier,那你知道非对齐的barrier吗?如何使用呢?让我们通过下面的阅读一起了解一下吧。
一、Barrier
流的barrier是Flink Checkpoint中的一个核心概念。多个barrier被插入到数据流中, 然后作为数据流的一部分随着数据流动(有点类似于Watermark)。这些barrier不会跨越流中的数据。
每个barrier会把数据流分成两部分:一部分数据进入当前的快照,另一部分数据进入下一个快照。每个barrier携带着快照的id。barrier 不会暂停数据的流动,所以非常轻量级。在流中, 同一时间可以有来源于多个不同快照的多个barrier,这意味着可以并发地出现不同的快照。
二、对齐的barrier
在多并行度下,如果要实现严格一次,则要执行barrier对齐。当 job graph 中的每个 operator 接收到 barrier 时,就会记录下其状态。拥有两个输入流的 operators(例如 CoProcessFunction)会执行 barrier 对齐(barrier alignment),以便当前快照能够包含两个输入流 barrier 之前(但不超过)的所有 events 产生的状态。
1. 当operator收到数字流的barrier n时, 它就不能处理(但是可以接收)来自该流的任何数据记录,直到它从字母流的所有输入中接收到 barrier n 为止。
2. 接收到 barrier n 的流(数字流)暂时被搁置。从这些流接收的记录会进入输入缓冲区, 不会被处理。例如图中的 barrier n 之后的数据 123 已经到达了operator, 存入到了输入缓冲区没有被处理, 只有等到字母流的 barrier n 到达之后才会开始处理。
3. 一旦最后所有输入流都接收到 barrier n,operator 就会把缓冲区中待输出的数据发出去,然后把 barrier n 接着往下游发送。这里还会对自身进行快照。
优点:
-
barrier 对齐不仅保证了状态的准确性,还巧妙地消去了原生C-L算法中记录输入流状态的步骤,十分轻量级,保存的数据体积小。
缺点:
-
延迟性高(快的barrier到达后会阻塞此条流的数据处理)。
-
加剧作业的反压(当出现反压时,数据本身就处理不过来,此时某条流的数据又阻塞了,所以就会加剧反压)。
-
整体 chenkpoint 时间变长(因为反压会导致数据流速变慢,导致barrier流动速度也会变慢,所以整体chenkpoint时间就会变长)。
三、barrier不对齐
如果barrier不对齐会怎么样?会重复消费,就是至少一次语义。
1. 当 operator 收到数字流的 barrier n 时,开启本地快照记录自己的状态,并将这个 barrier 发往下游(输出缓冲区)。
2. 接收到 barrier n 的流(数字流)会继续往下走。字母流的 barrier n 前面的数据(abcd)会被保存到状态里面,直到 barrier n 到来以后,再进行checkpoint,将数据保存到检查点中。
优点:
-
避免了 checkpoint 可能带来的阻塞,有利于提升 Flink 的资源利用率。
缺点:
-
由于要持久化缓存数据,State Size 会有比较大的增长,磁盘负载会加重。
-
随着 State Size 增长,作业恢复时间可能增长,运维管理难度增加。
四、barrier的使用
对齐
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 每 1000ms 开始一次 checkpoint
env.enableCheckpointing(1000);
// 高级选项:
// 设置模式为精确一次 (这是默认值),对于延迟要求较高的选择,最少一次
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 确认 checkpoints 之间的时间会进行 500 ms
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
// Checkpoint 必须在一分钟内完成,否则就会被抛弃
env.getCheckpointConfig().setCheckpointTimeout(60000);
// 允许两个连续的 checkpoint 错误
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
// 同一时间只允许一个 checkpoint 进行
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
// 使用 externalized checkpoints,这样 checkpoint 在作业取消后仍就会被保留
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
// 开启实验性的 unaligned checkpoints
env.getCheckpointConfig().enableUnalignedCheckpoints();
非对齐
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 启用非对齐 Checkpoint
env.getCheckpointConfig().enableUnalignedCheckpoints();
或者在 flink-conf.yml 配置文件中增加配置:
execution.checkpointing.unaligned: true
总结
非对齐barrier主要是解决严重反压情况下作业难以完成 checkpoint 的问题,同时它以磁盘资源为代价,避免了 checkpoint 可能带来的阻塞,有利于提升 Flink 的资源利用率。
更多推荐
所有评论(0)