flink1.11针对反压背压的优化

2020-10-22 本文已影响0人 feng504x

背景:

最近flink发布新版本1.11, 除了优化旧版本已有的缺陷, 还增加了一些新功能, 其中我发现有一些改变适合用于现在负责的flink项目

我们当前的flink项目的问题是生成checkpoint失败较多,造成checkpoint失败的原因是某几个subtask的快照超时导致整体的checkpoint生成失败,随着每天的处理的任务越多, 这个问题越发突显出来, 而后果是:

引用的答案:

目前的 Checkpoint 算法在大多数情况下运行良好，然而当作业出现反压时，阻塞式的 Barrier 对齐反而会加剧作业的反压，甚至导致作业的不稳定。首先， Chandy-Lamport 分布式快照的结束依赖于 Marker 的流动，而反压则会限制 Marker 的流动，导致快照的完成时间变长甚至超时。无论是哪种情况，都会导致 Checkpoint 的时间点落后于实际数据流较多。这时作业的计算进度是没有被持久化的，处于一个比较脆弱的状态，如果作业出于异常被动重启或者被用户主动重启，作业会回滚丢失一定的进度。如果 Checkpoint 连续超时且没有很好的监控，回滚丢失的进度可能高达一天以上，对于实时业务这通常是不可接受的。更糟糕的是，回滚后的作业落后的 Lag 更大，通常带来更大的反压，形成一个恶性循环。

优化1: rebalance分区改为rescale分区

rebalance使用Round-ribon思想将数据均匀分配到各实例上。Round-ribon是负载均衡领域经常使用的均匀分配的方法，上游的数据会轮询式地分配到下游的所有的实例上。如下图所示，上游的算子会将数据依次发送给下游所有算子实例。

rebalance分区

dataStream.rebalance()

rescale与rebalance很像，也是将数据均匀分布到各下游各实例上，但它的传输开销更小，因为rescale并不是将每个数据轮询地发送给下游每个实例，而是就近发送给下游实例。

dataStream.rescale()

rescale分区1

如上图所示，当上游有两个实例时，上游第一个实例将数据发送给下游第一个和第二个实例，上游第二个实例将数据发送给下游第三个和第四个实例，相比rebalance将数据发送给下游每个实例，rescale的传输开销更小。下图则展示了当上游有四个实例，上游前两个实例将数据发送给下游第一个实例，上游后两个实例将数据发送给下游第二个实例。

rescale分区2

优化2: 升级flink1.11, 使用Unaligned Checkpoint + rocksdb生成Checkpoint

flink1.11新特性相关介绍: https://www.h5w3.com/33867.html

Rocksdb state ssd: 使用rocksdb缓存checkpoint, 并且从原来的全量生成改为增量生成的方式, 速度更快

Unaligned Checkpoint

Flink 现有的 Checkpoint 机制下，每个算子需要等到收到所有上游发送的 Barrier 对齐后才可以进行 Snapshot 并继续向后发送 barrier。在反压的情况下，Barrier 从上游算子传送到下游可能需要很长的时间，从而导致 Checkpoint 超时的问题。

针对这一问题，Flink 1.11 增加了 Unaligned Checkpoint 机制。开启 Unaligned Checkpoint 后当收到第一个 barrier 时就可以执行 checkpoint，并把上下游之间正在传输的数据也作为状态保存到快照中，这样 checkpoint 的完成时间大大缩短，不再依赖于算子的处理能力，解决了反压场景下 checkpoint 长期做不出来的问题。

可以通过 env.getCheckpointConfig().enableUnalignedCheckpoints();开启unaligned Checkpoint 机制。