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

引用Apache Flink常见问题:

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

默认情况下,Flink占网络缓冲区JVM堆大小的10%,最小值为64MB,最大值为1GB。您可以通过taskmanager调整所有这些值。网络记忆力分数,taskmanager。网络记忆力min和taskmanager。网络记忆力最大值。

有关详细信息,请参阅配置参考。

有一个 dedicated section in the docs for how to configure the network buffers 。

总之,您可以在 ./conf/flink-conf.yaml 通过设置 taskmanager.network.numberOfBuffers 参数

参数应设置为 #slots-per-TM^2 * #TMs * 4 哪里 #slots per TM 是每个TaskManager的插槽数,以及 #TMs 是任务管理器的总数。

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

详情请参阅文件。

我也面临同样的错误

原因:java。io。IOException:网络数量不足 缓冲区:需要13个,但只有7个可用。的总数 网络缓冲区当前设置为2048,每个缓冲区32768字节。你可以 通过设置配置键来增加此数字 '任务管理器。记忆力网络分数', '任务管理器。记忆力网络最小值',和 '任务管理器。记忆力网络最大值'。

下面的代码片段解决了我的问题。

Configuration cfg = new Configuration();
int defaultLocalParallelism = Runtime.getRuntime().availableProcessors();
cfg.setString("taskmanager.memory.network.max", "1gb");
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(defaultLocalParallelism, cfg);