上接：flink 使用 ParameterTool 读取外部配置文件
说明
使用 ParameterTool 读取外部配置文件，在算子中访问配置，有下面两种方式
方式一
通过算子的构造方法，将 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 Std1 {
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(configTool))
.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>> {
private ParameterTool configTool;
public FlatMapFunImpl(ParameterTool configTool) {
this.configTool = configTool;
}
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
if(configTool.has("test")){
System.out.println(getClass().getName()+"配置文件参数："+configTool.get("test"));
}
String[] arr = value.split(",");
for(String item : arr){
Tuple2<String, Integer> temp = Tuple2.of(item, 1);
out.collect(temp);
}
}
}
}
测试
--config d:/flink.properties
方式二
将 ParameterTool 对象注册到 Flink 全局变量，在算子中使用
package std;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
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 Std2 {
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();
// 注册全局变量，在 Rich函数中可以获取
env.getConfig().setGlobalJobParameters(configTool);
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<>("kafka2flink", new SimpleStringSchema(), pro);
return fkc;
}
static class FlatMapFunImpl extends RichFlatMapFunction<String, Tuple2<String, Integer>> {
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
ParameterTool configTool = (ParameterTool)getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
if(configTool.has("test")){
System.out.println(getClass().getName()+"配置文件参数："+configTool.get("test"));
}
String[] arr = value.split(",");
for(String item : arr){
Tuple2<String, Integer> temp = Tuple2.of(item, 1);
out.collect(temp);
}
}
}
}