在Flink任务中，需要加载外置配置参数到任务中，在Flink的开发文档中介绍了，Flink提供了一个名为 ParameterTool 的工具来解决这个问题
Flink开发文档:
1 2 https://github.com/apache/flink/blob/master/docs/dev/application_parameters.zh.md
其引入配置的方式有3种：
1.From .properties files
1 2 3 String propertiesFilePath = "/home/sam/flink/myjob.properties"; ParameterTool parameter = ParameterTool.fromPropertiesFile(propertiesFilePath);
2. From the command line arguments
在args中添加
1 2 --input hdfs:///mydata --elements 42
在代码中使用
1 2 3 4 5 6 public static void main(String[] args) { // parse input arguments final ParameterTool parameterTool = ParameterTool.fromArgs(args); }
3. From system properties
使用
或者
1 2 ParameterTool parameter = ParameterTool.fromSystemProperties();
ParameterTool有如下几个方法可以获得参数
1 2 3 4 5 parameter.getRequired("input"); parameter.get("output", "myDefaultValue"); parameter.getLong("expectedCount", -1L); parameter.getNumberOfParameters();
注册全局变量
1 2 3 4 5 6 // parse input arguments ParameterTool parameters = ParameterTool.fromSystemProperties(); // register the parameters globally ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); env.getConfig().setGlobalJobParameters(parameters);
下面使用第1种方法来配置Flink消费Kafka的Topic
Ref:
1 2 https://github.com/apache/flink/blob/master/flink-end-to-end-tests/flink-streaming-kafka-test-base/src/main/java/org/apache/flink/streaming/kafka/test/base/KafkaExampleUtil.java
对于带有前缀的配置 读取，可以参考flume的前缀配置读取方法 getSubProperties()，使用其从整体的配置文件中读取 prefix 开头的配置，并去掉prefix
1 https://github.com/apache/flume/blob/trunk/flume-ng-configuration/src/main/java/org/apache/flume/Context.javab
通用配置读取工具类 ParameterToolUtil
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Maps; import java.util.Map; public class ParameterToolUtil { // 读取parameters中前缀为prefix的配置，并去掉前缀 public static Map<String, String> getSubProperties(Map<String, String> parameters, String prefix) { Preconditions.checkArgument(prefix.endsWith("."), "The given prefix does not end with a period (" + prefix + ")"); Map<String, String> result = Maps.newHashMap(); synchronized (parameters) { for (Map.Entry<String, String> entry : parameters.entrySet()) { String key = entry.getKey(); if (key.startsWith(prefix)) { String name = key.substring(prefix.length()); result.put(name, entry.getValue()); } } } return ImmutableMap.copyOf(result); } }
比如如下配置
1 2 3 4 5 # kafka source kafka.source.bootstrap.servers=localhost:9092 kafka.source.topics=thrift_log_test kafka.source.group.id=test
可以这样读取
1 2 3 4 // 读取kafka相关的配置参数 Properties kafkaSourceProps = new Properties(); kafkaSourceProps.putAll(ParameterToolUtil.getSubProperties(parameters, KafkaSourceConstants.KAFKA_SOURCE_PREFIX));
其中 KafkaSourceConstants.KAFKA_SOURCE_PREFIX 常量为 kafka.source.
使用 getSubProperties() 方法读取的时候将会读取带有 prefix 前缀的配置，并自动去掉前缀
