Flink传入自定义的参数或配置文件

一、Flink 动态参数传入方式简介

Flink提供ParameterTool类，从不同的来源读取参数

1.fromArgs方法
2.fromPropertiesFile方法
3.fromSystemProperties方法

二、Flink 动态参数实操

1.fromArgs方法

读取在命令行中传递的参数，注意传参 格式为 key value ，key必须以 - 或者 – 开头，如 --key1 value1 --key2 value2 -key3 value3

eg: 传参格式为 --type:“stock” --markType:“101” 空格间隔

• 动态收参代码

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 获取传入的参数
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        // 注册给环境变量
        env.getConfig().setGlobalJobParameters(parameterTool);
        // 获取注册的配置
        ExecutionConfig.GlobalJobParameters parameters = env.getConfig().getGlobalJobParameters();
        Map<String, String> map = parameters.toMap();
        String markType = map.get("markType");

        // 设置任务失败重启 允许任务失败最大次数 3次
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3,
                // 任务失败的时间启动的间隔
                Time.of(2, TimeUnit.SECONDS),
                // 允许任务延迟时间 3s
                Time.of(3, TimeUnit.SECONDS))
        );

2.fromPropertiesFile方法

• 配置文件Flink-conf.yarml编写方式

#hbase连接地址
hbase_zookeeper_quorum: localhost
hbase_zookeeper_client_port: 2181

#kafka连接地址
kafka_source_quorum: 
kafka_source_topic: 
kafka_source_group: 

记得yml文件 冒号":" 和地址 中间加空格哦！

• 动态收参代码

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		ParameterTool parameterTool = ParameterTool.fromPropertiesFile("配置文件目录");
        // 注册给环境变量
        env.getConfig().setGlobalJobParameters(parameterTool);
        // 获取注册的配置
        ExecutionConfig.GlobalJobParameters parameters = env.getConfig().getGlobalJobParameters();
		//设置静态类调用
        new FlinkConf(parameters);

public class FlinkConf {
	//静态类
	public static String HBASE_SERVER ;
    public static String KAFKA_SERVER ;
	public FlinkConf(ExecutionConfig.GlobalJobParameters parameters){
        Map<String, String> map = parameters.toMap();
        HBASE_SERVER = map.get("hbase_zookeeper_quorum");
        KAFKA_SERVER = map.get("kafka_quorum");
    }
}