Flink IOException：网络缓冲区数量不足

Question

我正在使用Flink v1.4.0. 我正在使用DataSet API（虽然这个，我认为无关紧要）。

我正在 12 核 VM 上运行一些重型转换。我正在使用 2 个内核Flink job，其中我将一些数据存储到一个内核中，并使用剩余的 10 个内核Flink Queryable State运行另一个Flink作业。

当我用 10 个内核运行第二个作业时，我似乎收到以下错误：

java.io.IOException: Insufficient number of network buffers: required 10, but only 9 available. The total number of network buffers is currently set to 4096 of 32768 bytes each. You can increase this number by setting the configuration keys 'taskmanager.network.memory.fraction', 'taskmanager.network.memory.min', and 'taskmanager.network.memory.max'.
            at org.apache.flink.runtime.io.network.buffer.NetworkBufferPool.createBufferPool(NetworkBufferPool.java:257)
            at org.apache.flink.runtime.io.network.NetworkEnvironment.registerTask(NetworkEnvironment.java:199)
            at org.apache.flink.runtime.taskmanager.Task.run(Task.java:618)
            at java.lang.Thread.run(Thread.java:745)

如果我确实用 8 个内核运行它，它就可以正常运行。这是什么原因造成的，为什么我不能使用其他 2 --> 8+2 = 10 个内核？

Answer 1

引用 Apache Flink 常见问题解答：

如果以非常高的并行度运行 Flink，则可能需要增加网络缓冲区的数量。

默认情况下，Flink 占用 JVM 堆大小的 10% 用于网络缓冲区，最小为 64MB，最大为 1GB。您可以通过 taskmanager.network.memory.fraction、taskmanager.network.memory.min 和 taskmanager.network.memory.max 调整所有这些值。

详情请参考配置参考。

文档中有专门的部分介绍如何配置网络缓冲区。

综上所述，您可以./conf/flink-conf.yaml通过设置taskmanager.network.numberOfBuffers参数来配置文件中网络缓冲区的数量。

该参数应设置为#slots-per-TM^2 * #TMs * 4，其中#slots per TM是每个 TaskManager 的插槽数和#TMs任务管理器的总数。

例如，为了支持 20 台 8 槽机器的集群，您应该使用大约 5000 个网络缓冲区以获得最佳吞吐量。默认情况下，每个网络缓冲区的大小为 32 KB。在上面的示例中，系统因此将为网络缓冲区分配大约 300 MiBytes。

有关详细信息，请参阅文档。