上接：flink 使用 ParameterTool 解析命令行参数
说明
在Flink中，一般从 Kafka 读取数据，计算后，可能将结果保存到 hbase、elastic search 等外部存储介质中
与 springboot 工程类似，需要将连接参数保存到配置文件中，而不是把参数硬编码在代码中
可以通过以下两种方式实现
方式一
参数较少，直接在启动命令后面加参数，然后用 ParameterTool 解析，详见下面链接：
flink 使用 ParameterTool 解析命令行参数
方式二
参数较多，将参数配置到 .properties 文件中，在启动命令后面加 该文件路径的参数，然后用 ParameterTool 解析
例子
package std;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Properties;
/**
* 运行时需要指定参数：--config d:/flink.properties
*/
public class ReadConfig {
public static void main(String[] args) throws Exception {
ParameterTool pt = ParameterTool.fromArgs(args);
if(!pt.has("config")){
throw new FileNotFoundException("没有指定 config 配置文件");
}
ParameterTool configTool = ParameterTool.fromPropertiesFile(pt.get("config"));
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 处理
SingleOutputStreamOperator<Tuple2<String, Integer>> soso = env.addSource(buildKafka(configTool))
.flatMap(new FlatMapFunImpl())
.keyBy(value -> value.f0)
.sum(1);
soso.print("res:");
env.execute("myjob");
}
private static FlinkKafkaConsumer<String> buildKafka(ParameterTool configTool) throws IOException {
Properties pro = new Properties();
if(configTool.has(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configTool.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
}
if(configTool.has(ConsumerConfig.GROUP_ID_CONFIG)) {
pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG, configTool.get(ConsumerConfig.GROUP_ID_CONFIG));
}
if(configTool.has(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)) {
pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, configTool.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG));
}
// 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>(configTool.get("topic"), new SimpleStringSchema(), pro);
return fkc;
}
static class FlatMapFunImpl implements FlatMapFunction<String, Tuple2<String, Integer>> {
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
String[] arr = value.split(",");
for(String item : arr){
Tuple2<String, Integer> temp = Tuple2.of(item, 1);
out.collect(temp);
}
}
}
}
idea 配置如下：
--config d:/flink.properties