Flink 网络缓冲区使用率高，导致 Kafka 滞后答案

【问题标题】：High Flink network buffer usage, which causes Kafka laggingFlink 网络缓冲区使用率高，导致 Kafka 滞后
【发布时间】：2021-01-30 13:40:05
【问题描述】：

我们的 Flink Jobs 包含一个过滤器，按会话 id 键和间隔 30 分钟的会话窗口。会话窗口需要累积会话的所有事件，并使用ProcessWindowFunction 处理它们。

我们正在使用 Flink 1.9，总共 20G 内存的 128 个容器来运行我们的作业，截止比率为 0.3。我们正在做增量检查点。

当会话窗口开始触发process 函数时，网络缓冲区使用率开始变得相当高，然后我们开始出现 Kafka 输入滞后。我们的设置：

state.backend: rocksdb
state.checkpoints.dir: hdfs://nameservice0/service
state.backend.rocksdb.memory.managed: true
state.backend.incremental: true
#https://github.com/facebook/rocksdb/wiki/Memory-usage-in-RocksDB
state.backend.rocksdb.memory.write-buffer-ratio: 0.6
state.backend.rocksdb.memory.high-prio-pool-ratio: 0.1
state.backend.rocksdb.block.blocksize: 16mb
state.backend.rocksdb.writebuffer.count: 8
state.backend.rocksdb.writebuffer.size: 256mb
state.backend.rocksdb.timer-service.factory: heap

containerized.heap-cutoff-ratio: 0.25
taskmanager.network.memory.fraction: 0.85
taskmanager.network.memory.min: 512mb
taskmanager.network.memory.max: 7168mb
taskmanager.network.memory.buffers-per-channel: 8
taskmanager.memory.segment-size: 4mb
taskmanager.network.memory.floating-buffers-per-gate: 16
taskmanager.network.netty.transport: poll

部分图表：

任何建议将不胜感激！

【问题讨论】：

如果您从 kafka 代理消费，我会首先检查您的 kafka 源是否从主题开始消费消息。也可能是您的 kafka 代理有很多消息/事件等待消费。之后检查检查点指标是有意义的
@Felipe，谢谢。每次开始测试之前，我都会将 Kafka 偏移量重置为最新的

标签： apache-flink rocksdb flink-cep flink-statefun

【解决方案1】：

我不知道 flink 的内部结构，但原因可能与会话窗口有关。我的意思是，如果您有这么多会话操作具有相同的时间间隔（30 分钟），所有会话操作将同时执行，这会造成延迟。

【讨论】：

【解决方案2】：

如果我可以访问详细信息，我会通过以下方式尝试提高此应用程序的性能：

(1) 是否可以重新实现窗口以进行增量聚合？目前，这些窗口正在构建可能相当长的事件列表，并且它们仅在会话结束时通过这些列表工作。这显然需要足够长的时间才能对 Kafka 造成背压。如果您可以预先聚合会话结果，这将平衡处理，并且问题应该会消失。

不，我并没有与我所说的 here 相矛盾。如果我不清楚，请告诉我。

(2) 你已经放置了很多额外的网络缓冲。这通常会适得其反；您希望背压快速反弹并限制源，而不是将更多数据推入 Flink 的网络缓冲区。

您最好减少网络缓冲，如果可能，请使用您的可用资源来提供更多插槽。当一个插槽忙于处理刚刚结束的长会话的内容时，拥有更多插槽将减少影响。为 RocksDB 提供更多内存可能也会有所帮助。

(3) 看看能不能优化序列化。最好和最差的串行器之间的吞吐量可能相差 10 倍。见Flink Serialization Tuning。如果记录中有您实际上不需要的字段，请将它们删除。

(4) 看看调优 RocksDB。确保为 RocksDB 使用最快的可用本地磁盘，例如本地 SSD。避免为state.backend.rocksdb.localdir 使用网络附加存储（例如EBS）。

【讨论】：