研究內容

flink客戶端提交命令為 flink run ....

如果客戶端的main 需要讀取系統變數。

讀取系統變數的位置有兩種：

• 從作業的main方法中讀取。
• 從作業的運算元中讀取。

測試環境

Flink -m yarn-cluster

測試方法

對於JVM引數指定的系統變數

-Dkafka.start_from_timestamp=1648828800000

指定方式

FLINK_ENV_JAVA_OPTS="-Dkafka.start_from_timestamp=1648828800009" bin/flink run ...

讀取環境變數

String property = System.getProperty("kafka.start_from_timestamp");
if (property == null) {
    //-Dkafka.start_from_timestamp=1648828800000
    System.err.println("-Dkafka.start_from_timestamp Not found");
    System.err.println("This are Properties Found in this JVM:");
    System.err.println(System.getProperties().stringPropertyNames());
} else {
    System.err.println("-Dkafka.start_from_timestamp is" + property);
}
 

• 讀取位置1（在運算元外部）提交日誌中的輸出：
  -Dkafka.start_from_timestamp is1648828800009

• 讀取位置2（在運算元內部）TaskManager 輸出結果：

-Dkafka.start_from_timestamp Not found
This are Properties Found in this JVM:
[zookeeper.sasl.client, java.runtime.name, sun.boot.library.path, java.vm.version, java.vm.vendor, java.vendor.url, path.separator, java.vm.name, file.encoding.pkg, user.country, sun.java.launcher, sun.os.patch.level, java.vm.specification.name, user.dir, java.runtime.version, java.awt.graphicsenv, java.endorsed.dirs, os.arch, java.io.tmpdir, line.separator, java.vm.specification.vendor, os.name, log4j.configuration, sun.jnu.encoding, java.library.path, sun.nio.ch.bugLevel, java.specification.name, java.class.version, sun.management.compiler, os.version, user.home, user.timezone, java.awt.printerjob, file.encoding, java.specification.version, log4j.configurationFile, user.name, java.class.path, log.file, java.vm.specification.version, sun.arch.data.model, java.home, sun.java.command, java.specification.vendor, user.language, awt.toolkit, java.vm.info, java.version, java.ext.dirs, sun.boot.class.path, java.vendor, java.security.auth.login.config, file.separator, java.vendor.url.bug, sun.cpu.endian, sun.io.unicode.encoding, sun.cpu.isalist]
 

測試專案

1. flink-conf.yaml 中指定 env.java.opts

2. FLINK_ENV_JAVA_OPTS 指定 -Dkey=value這樣的 System Properties

3. 在 flink run -m yarn-cluster ... -yD env.java.opts="自定義引數" 中這樣指定

注：env.java.opts.client, env.java.opts.taskmanager 的測試方法均類似

本文僅以第二種指定方式為例。其他方式不做贅述。

結論

• 對於 FLINK_ENV_JAVA_OPTS 系統環境變數設定的自定義系統變數，僅在客戶端提交作業過程中可以訪問。
• 對於 flink-conf.yaml 中的 env.java.opts 經過驗證，可以在客戶端以及TaskManager中訪問。
• 對於 flink run -m yarn-cluster ... -yD env.java.opts="自定義引數"，僅在運算元也就是在TaskManager中可以訪問，客戶端中無法訪問。

參考連結

FLINK-27130