Flink:执行管道错误java.util.concurrent.TimeoutException:期货超时后[10000毫秒]

Question

我正在使用Flink v1.4.0。

我正在利用Flink的本地图形API (Gelly)来处理1200万个数据点(边缘)。我目前正在使用IntelliJ运行我的作业，它在同一个JVM中执行所有的TaskManagers和JobManager。

在加载数据、有效地生成边缘时，在运行Flink作业之前，总是会出现以下异常：

...
Connected to JobManager at Actor[akka://flink/user/jobmanager_1#XXXXXXXXXX] with leader session id XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX.
322062 [main] ERROR com.somecompany.some.domain.name.some.javaClass- Error executing pipeline
java.util.concurrent.TimeoutException: Futures timed out after [10000 milliseconds]
            at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:223)
            at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:157)
            at scala.concurrent.Await$$anonfun$ready$1.apply(package.scala:169)
            at scala.concurrent.Await$$anonfun$ready$1.apply(package.scala:169)
            at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53)

..。

我确保通过IntelliJ编辑运行配置以添加：

-Dakka.client.timeout:600s
-Dakka.ask.timeout:600s

但例外情况依然存在，我不知道是什么原因造成的。当我缩小数据大小时，一切都很好。

当我试图通过Flink UI提交相同的作业以在集群上安装的本地Flink版本上运行时，也会出现同样的问题。在这种情况下，作业从未开始，我甚至无法预览自动生成的操作符DAG。同样，当我减少要处理的数据量时，问题就消失了。我还更新了flink-conf.yaml，使其包含相同的akka配置属性，但这并没有什么区别。

我该怎么解决这个问题？

Answer 1

在IntelliJ中运行Flink作业时，依赖于Flink迷你集群。mini集群与在独立的Yarn或Mesos上运行Flink不同，因为它依赖于单个JVM。此外，迷你集群在很多方面都是预先配置的，而改变配置并不总是可能的(至少在某些设置方面是如此)。

当我向集群提交作业时(而不是在通过迷你集群运行作业时)，我必须更改的一件事是我分配给作业管理器的堆内存的大小。这是必要的，因为加载要处理的数据不是我想要运行的Flink作业的一部分(这不是Flink中的标准实践，这样做实际上是错误的)。通过增加作业管理器的堆，我能够让我的作业运行，但最终我不得不为我的Flink作业定义一个新的输入格式，以避免Job不得不预加载数据以执行--无论如何，这不应该是它的责任。

对于眼前的问题:不能通过IntelliJ完成堆内存分配给作业管理器(据我所知)，因此作业总是失败的。