Flink最佳实践

╰+哭是因爲堅強的太久メ 2022-12-01 11:54 326阅读 0赞

## 一.传递参数 ##

几乎所有的Flink应用程序（批处理和流式处理）都依赖于外部配置参数。它们用于指定输入和输出源（如路径或地址），系统参数（并行度，运行时配置）和特定于应用程序的参数（通常在用户功能内使用）。

Flink提供了一个简单的实用程序，称为ParameterTool提供解决这些问题的一些基本工具。请注意，使用ParameterTool不是必须的。其他框架（例如Commons CLI和 argparse4j）也可以与Flink一起使用。

**将配置值放入ParameterTool**  
所述ParameterTool用于读取所述配置提供一组预定义的静态方法。该工具在内部期望使用Map<String, String>，因此将其与自己的配置样式集成起来非常容易。  
**从配置文件中加载参数**：

//url
    String propertiesFilePath = "/home/sam/flink/myjob.properties";
    ParameterTool parameter = ParameterTool.fromPropertiesFile(propertiesFilePath);
    //file
    File propertiesFile = new File(propertiesFilePath);
    ParameterTool parameter = ParameterTool.fromPropertiesFile(propertiesFile);
    //fileInputStream
    InputStream propertiesFileInputStream = new FileInputStream(file);
    ParameterTool parameter = ParameterTool.fromPropertiesFile(propertiesFileInputStream);

**从命令行加载参数**  
这样可以–input hdfs:///mydata --elements 42从命令行获取参数。

public static void main(String[] args) { 
        ParameterTool parameter = ParameterTool.fromArgs(args);

**从系统属性加载参数**  
启动JVM时，可以将系统属性传递给它：-Dinput=hdfs:///mydata。还可以通过使用ParameterTool加载系统属性初始化：

ParameterTool parameter = ParameterTool.fromSystemProperties();

## 二.获取参数 ##

使用ParameterTool本身访问值的方法。

ParameterTool parameters = // ...
    parameter.getRequired("input");
    parameter.get("output", "myDefaultValue");
    parameter.getLong("expectedCount", -1L);
    parameter.getNumberOfParameters()

可以直接在main()方法提交应用程序的客户端方法中使用这些方法的返回值。例如，可以像这样设置运算符的并行性：

ParameterTool parameters = ParameterTool.fromArgs(args);
    int parallelism = parameters.get("mapParallelism", 2);
    DataSet<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer()).setParallelism(parallelism);

由于ParameterTool可以序列化，因此可以将其传递给函数本身：

ParameterTool parameters = ParameterTool.fromArgs(args);
    DataSet<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer(parameters));

然后在函数内部使用它从命令行获取值。  
**全局注册参数**  
ExecutionConfig可以从JobManager Web界面以及用户定义的所有功能中将其作为配置值注册为全局作业参数。

全局注册参数：

ParameterTool parameters = ParameterTool.fromArgs(args);
    
    // set up the execution environment
    final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    env.getConfig().setGlobalJobParameters(parameters);

通过任何丰富的用户功能访问它们：

public static final class Tokenizer extends RichFlatMapFunction<String, Tuple2<String, Integer>> { 
    
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) { 
    	ParameterTool parameters = (ParameterTool)
    	    getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
    	parameters.getRequired("input");

## 三.命名大型TupleX类型 ##

建议使用POJO（普通的旧Java对象），而不要TupleX使用具有许多字段的数据类型。同样，POJO可以用来给大型Tuple类型起一个名字。

Tuple11<String, String, ..., String> var = new ...;

从大型Tuple类型创建扩展的自定义类型要容易得多。

CustomType var = new ...;
    
    public static class CustomType extends Tuple11<String, String, ..., String> { 
        // constructor matching super
    }

## 四.使用Logback代替Log4j ##

注意：从Flink 0.10开始适用

Apache Flink使用slf4j作为代码中的日志记录抽象。建议用户在其用户功能中也使用sfl4j。

Sfl4j是一个编译时日志记录接口，可以在运行时使用不同的日志记录实现，例如log4j或Logback。

要在代码中获取记录器实例，请使用以下代码：

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    public class MyClass implements MapFunction { 
        private static final Logger LOG = LoggerFactory.getLogger(MyClass.class);