Flink passes in custom parameters or profiles

One 、Flink Introduction to the dynamic parameter input method

Flink Provide ParameterTool class , Read parameters from different sources

1.fromArgs Method
2.fromPropertiesFile Method
3.fromSystemProperties Method

Two 、Flink Dynamic parameter practice

1.fromArgs Method

Read the parameters passed on the command line , Pay attention to the transmission of ginseng The format is key value ,key Must be - perhaps – start , Such as --key1 value1 --key2 value2 -key3 value3

eg: The format of parameter transfer is --type:“stock” --markType:“101” Space interval

• Dynamic parameter receiving code

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //  Get the parameters passed in 
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        //  Register to environment variables 
        env.getConfig().setGlobalJobParameters(parameterTool);
        //  Get the registered configuration 
        ExecutionConfig.GlobalJobParameters parameters = env.getConfig().getGlobalJobParameters();
        Map<String, String> map = parameters.toMap();
        String markType = map.get("markType");

        //  Failed to set task restart   Maximum number of task failures allowed  3 Time 
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3,
                //  The time when the task failed and the interval between starts 
                Time.of(2, TimeUnit.SECONDS),
                //  Allow task delay time  3s
                Time.of(3, TimeUnit.SECONDS))
        );

2.fromPropertiesFile Method

• The configuration file Flink-conf.yarml Compiling mode

#hbase Connection address 
hbase_zookeeper_quorum: localhost
hbase_zookeeper_client_port: 2181

#kafka Connection address 
kafka_source_quorum: 
kafka_source_topic: 
kafka_source_group: 

Remember yml file The colon ":" And address Add a space in the middle ！

• Dynamic parameter receiving code

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		ParameterTool parameterTool = ParameterTool.fromPropertiesFile(" Profile directory ");
        //  Register to environment variables 
        env.getConfig().setGlobalJobParameters(parameterTool);
        //  Get the registered configuration 
        ExecutionConfig.GlobalJobParameters parameters = env.getConfig().getGlobalJobParameters();
		// Set static class calls 
        new FlinkConf(parameters);

public class FlinkConf {
	// Static class 
	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");
    }
}