Flink网络缓冲区使用率较高，这会导致Kafka滞后

如何解决Flink网络缓冲区使用率较高，这会导致Kafka滞后

我们的Flink作业包含一个过滤器，按会话ID键，然后按间隔30分钟的会话窗口进行过滤。会话窗口将需要累积会话的所有事件，并使用ProcessWindowFunction处理它们。

我们使用Flink 1.9，总共128个具有20G内存的容器来运行我们的工作，截止率为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

一些图：

任何建议将不胜感激！

解决方法

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

,

如果我可以访问详细信息，则可以尝试以下方法来提高此应用程序的性能：

（1）是否可以重新实现Windows以进行增量聚合？当前，窗口正在建立可能是相当长的事件列表，并且仅在会话结束时才通过这些列表进行工作。显然，这花费了足够长的时间，从而导致卡夫卡背压。如果您可以预先汇总会话结果，则这将使处理过程变得均匀，并且问题将消失。

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

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

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

（3）看看是否可以优化序列化。最佳和最差的串行器之间的吞吐量可能相差10倍。参见Flink Serialization Tuning。如果记录中确实不需要任何字段，请将其删除。

（4）查看调整RocksDB。确保您使用的是RocksDB最快可用的本地磁盘，例如本地SSD。避免将网络附加存储（例如EBS）用于state.backend.rocksdb.localdir。