从YARN的另一个程序开始的Flink工作失败,"JobClientActor似乎已经死了"

Question

我是新的flink用户,我有以下问题.我在YARN集群上使用flink将从RDBMS提取的相关数据传输到HBase.我在具有多个ExecutionEnvironments(每个RDB表一个以并行传输表行)的java上编写flink批处理应用程序,以按顺序传递表(因为env.execute()的调用是阻塞的).

我像这样开始YARN会话

export YARN_CONF_DIR=/etc/hadoop/conf
export FLINK_HOME=/opt/flink-1.3.1
export FLINK_CONF_DIR=$FLINK_HOME/conf
$FLINK_HOME/bin/yarn-session.sh -n 1 -s 4 -d -jm 2048 -tm 8096

然后我在通过shell脚本transfer.sh启动的YARN会话上运行我的应用程序.它的内容就在这里

#!/bin/bash

export YARN_CONF_DIR=/etc/hadoop/conf
export FLINK_HOME=/opt/flink-1.3.1
export FLINK_CONF_DIR=$FLINK_HOME/conf
$FLINK_HOME/bin/flink run -p 4 transfer.jar

当我从命令行手动启动此脚本时,它工作正常 - 作业逐个提交给YARN会话而没有错误.

现在我应该能够从另一个java程序运行这个脚本.为了这个目的,我使用

Runtime.exec("transfer.sh");

(也许有更好的方法可以做到这一点？我已经在REST API中看到过,但由于作业管理器由YARN代理,因此存在一些困难).一开始就像往常一样工作 - 首先将几个工作提交到会话并成功完成.但以下工作未提交给YARN会议.在/opt/flink-1.3.1/log/flink-tsvetkoff-client-hadoop-dev1.log中我看到错误(并且在DEBUG级别中找不到其他错误)

The program execution failed: JobClientActor seems to have died before the JobExecutionResult could be retrieved.

我试图自己分析这个问题,发现在向JobClientActor(即YARN集群)发送带超时的ping请求时,JobClient类中发生了这个错误.我尝试增加多个心跳和超时选项,如akka.*.timeout,akka.watch.heartbeat.*和yarn.heartbeat-delay选项,但它没有解决问题 - 新的作业不会从CliFrontend提交给YARN会话.

两种情况(手动呼叫和来自另一个程序的呼叫)的环境是相同的.我打电话的时候

$ ps axu | grep transfer

它会给我输出

/usr/lib/jvm/java-8-oracle/bin/java -Dlog.file=/opt/flink-1.3.1/log/flink-tsvetkoff-client-hadoop-dev1.log -Dlog4j.configuration=file:/opt/flink-1.3.1/conf/log4j-cli.properties -Dlogback.configurationFile=file:/opt/flink-1.3.1/conf/logback.xml -classpath /opt/flink-1.3.1/lib/flink-metrics-graphite-1.3.1.jar:/opt/flink-1.3.1/lib/flink-python_2.11-1.3.1.jar:/opt/flink-1.3.1/lib/flink-shaded-hadoop2-uber-1.3.1.jar:/opt/flink-1.3.1/lib/log4j-1.2.17.jar:/opt/flink-1.3.1/lib/slf4j-log4j12-1.7.7.jar:/opt/flink-1.3.1/lib/flink-dist_2.11-1.3.1.jar:::/etc/hadoop/conf org.apache.flink.client.CliFrontend run -p 4 transfer.jar

我还尝试将flink更新到1.4.0版本或更改作业的并行性(甚至更改为-p 1)但仍然出现错误.

我不知道有什么不同？顺便说一下有没有解决方法？

感谢您的任何帮助.

Answer 1

最后我找到了如何解决该错误只需替换Runtime.exec(...)为new ProcessBuilder(...).inheritIO().start().

我真的不知道为什么在这种情况下调用inheritIOhelp ，因为据我了解，它只是将 IO 流从子进程重定向到父进程。但我检查过，如果我注释掉这行代码，程序就会再次开始崩溃。