Flink 检查点失败

Question

Flink 检查点失败

4

我们正在尝试使用RocksDB后端设置Flink有状态作业。我们使用30分钟的会话窗口，并使用聚合函数，因此不使用任何Flink状态变量。通过采样，我们每秒少于20k个事件，有20-30个新会话/秒。我们的会话基本上收集所有事件。随着时间的推移，会话累加器的大小会增加。我们在Flink 1.9中使用了总共10G内存，128个容器。以下是设置:

state.backend: rocksdb
state.checkpoints.dir: hdfs://nameservice0/myjob/path
state.backend.rocksdb.memory.managed: true
state.backend.incremental: true
state.backend.rocksdb.memory.write-buffer-ratio: 0.4
state.backend.rocksdb.memory.high-prio-pool-ratio: 0.1

containerized.heap-cutoff-ratio: 0.45
taskmanager.network.memory.fraction: 0.5
taskmanager.network.memory.min: 512mb
taskmanager.network.memory.max: 2560mb

在我们对一段时间的监控中，rocksdb内存表的大小小于10m，我们的堆使用量小于1G，但是我们的直接内存使用量（网络缓冲区）达到了2.5G。缓冲池/缓冲区使用度量都为1（满）。我们的检查点一直失败，我想知道网络缓冲区部分使用这么多内存是否正常？如果您能提出一些建议，我将不胜感激:)谢谢！

- lucky_start_izumi

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- David Anderson · Accepted Answer

就技术而言，会话窗口确实在内部使用Flink状态。（大多数来源和汇聚都是如此。）根据您将会话事件收集到会话累加器的方式，这可能是性能问题。如果您需要将所有事件汇总在一起，为什么不使用AggregateFunction让Flink代替您完成呢？

为获得最佳的窗口性能，您要使用ReduceFunction或AggregateFunction来逐渐减少/聚合窗口，仅保留最终成为窗口结果的一小部分状态。另一方面，如果您仅使用ProcessWindowFunction没有进行预聚合，那么Flink将在内部使用追加列表状态对象。当与RocksDB一起使用时，它非常高效，只需将每个事件序列化以将其附加到列表末尾。当最终触发窗口时，列表将作为可迭代对象传递给您，并被划分成块进行反序列化。另一方面，如果您使用AggregateFunction自己的解决方案，则可能导致RocksDB在每次访问/更新时进行反序列化和重新序列化。这可能非常昂贵，并且可能解释了检查点失败的原因。

您分享的另一个有趣事实是缓冲池/缓冲使用情况指标显示它们已完全利用。这表明存在显着的背压，这反过来将解释为什么检查点失败。检查点依赖于检查点障碍能够遍历整个执行图，检查点每个运算符，并在完成作业的完整扫描之前计时超时。有了背压，这可能会失败。

背压最常见的原因是低配-换句话说，集群被压垮了。网络缓冲池变得完全利用，因为运算符跟不上。答案不是增加缓冲区，而是去除/修复瓶颈。