Apache Flink-自定义Java选项在作业内未识别

Question

我已经在flink-conf.yaml中添加了以下行：

env.java.opts：" -ddy.props.path =/path/to/props/file"

启动JobManager（JobManager.sh启动群集）时，我在日志中看到JVM选项确实已识别

2017-02-20 12:19:23,536 INFO  org.apache.flink.runtime.jobmanager.JobManager                -  JVM Options:
2017-02-20 12:19:23,536 INFO  org.apache.flink.runtime.jobmanager.JobManager                -     -Xms256m
2017-02-20 12:19:23,536 INFO  org.apache.flink.runtime.jobmanager.JobManager                -     -Xmx256m
2017-02-20 12:19:23,536 INFO  org.apache.flink.runtime.jobmanager.JobManager                -     -XX:MaxPermSize=256m
2017-02-20 12:19:23,536 INFO  org.apache.flink.runtime.jobmanager.JobManager                -     -Ddy.props.path=/srv/dy/stream-aggregators/aggregators.conf
2017-02-20 12:19:23,536 INFO  org.apache.flink.runtime.jobmanager.JobManager                -     -Dlog.file=/srv/flink-1.2.0/log/flink-flink-jobmanager-0-flinkvm-master.log
2017-02-20 12:19:23,536 INFO  org.apache.flink.runtime.jobmanager.JobManager                -     -Dlog4j.configuration=file:/srv/flink-1.2.0/conf/log4j.properties
2017-02-20 12:19:23,536 INFO  org.apache.flink.runtime.jobmanager.JobManager                -     -Dlogback.configurationFile=file:/srv/flink-1.2.0/conf/logback.xml

但是，当我运行Flink Job（Flink Run -d Prog.jar）时，System.GetProperty（" dy.props.path"）会返回null（并且在打印系统属性时，我发现它确实不存在。）

问题确实是 - 如何设置将在Flink -Job代码中可用的系统属性？

Answer 1

这个问题与Flink的运行时架构非常相关[1]。

我知道您在独立的群集中运行工作。请记住，JobManager和TaskManager S在单独的JVM实例中运行。您必须考虑执行每个代码的何处。

例如，在TaskManager上执行了map或filter等转换中的代码。输入类的main方法中的代码是在命令行工具flink中执行的，该工具当然没有System属性集，因为它仅用于提交工作，它催生了临时（-D）JVM。

如果您通过WebUI提交工作，则在JobManager上执行main方法的代码，以便将属性设置为。

一般而言，我宁愿劝阻程序通过系统属性进行参数，因为这是一种不良习惯。

---

您有一个简单的示例：

我开始：

• 带有env.java.opts:"-Ddy.props.path=jobmanager"的JobManager
• TaskManager带有env.java.opts:"-Ddy.props.path=taskmanager"

我的工作代码如下：

object Main {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream = env.fromCollection(1 to 4)
    val prop = System.getProperty("dy.props.path")
    stream.map(_ => System.getProperty("dy.props.path") + "  mainArg: " + prop).print()
    env.execute("stream")
  }
}

当我通过flink工具提交代码时，输出如下：

taskmanager  mainArg: null
taskmanager  mainArg: null
taskmanager  mainArg: null
taskmanager  mainArg: null

当它通过WebUI提交时，我会得到：

taskmanager  mainArg: jobmanager
taskmanager  mainArg: jobmanager
taskmanager  mainArg: jobmanager
taskmanager  mainArg: jobmanager