Flink检查点不断失败

Question

我们正在尝试使用RocksDB后端设置Flink有状态作业。我们使用会话窗口，有30分钟的间隔。我们使用aggregateFunction，所以不使用任何Flink状态html" target="_blank">变量。通过采样，我们的事件数不到20k次/秒，新会话数不到20-30次/秒。我们的会议基本上收集了所有的事件。会话累加器的大小会随着时间而增大。我们总共使用了10G内存和Flink1.9，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（满）。我们的检查点总是失败，我想知道网络缓冲区部分可能会占用这么多内存是正常的吗？

Answer 1

值得注意的是，会话窗口确实在内部使用Flink状态。（大多数源和汇也是如此。）根据您将会话事件收集到会话累加器中的方式，这可能是一个性能问题。如果您需要将所有事件聚集在一起，为什么要使用AggregateFunction来完成这一操作，而不是让Flink为您完成这一操作？

为了获得最佳的窗口性能，您希望使用ReduceFunction或AggregateFunction来递增地缩小/聚合窗口，只保留一小部分最终将成为窗口结果的状态。另一方面，如果您只使用ProcessWindowFunction，而不使用预聚合，那么Flink将在内部使用一个附加列表状态对象，该对象与RocksDB一起使用时非常有效--它只需要序列化每个事件以将其附加到列表的末尾。当窗口最终被触发时，列表将作为一个以块形式反序列化的可迭代文件传递给您。另一方面，如果您使用AggregateFunction滚动您自己的解决方案，您可能会让RocksDB在每次访问/更新时反序列化和保留累加器。这可能会变得非常昂贵，并可能解释为什么检查点正在失败。

您分享的另一个有趣的事实是，缓冲池/缓冲区使用度量表明它们已被充分利用。这表明存在着巨大的背压，这反过来也解释了为什么检查点会失败。检查点依赖于检查点屏障能够遍历整个执行图，在每个操作符执行时对它们进行检查，并在超时前完成作业的全部扫描。如果有背压，这可能会失败。