Apache Flink API

骑猪看日落 2023-05-22 08:50 60阅读 0赞

**一、Apache Flink API**

Flink 提供了不同级别的抽象来开发流 / 批处理应用程序。![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNDQ3MzAx_size_16_color_FFFFFF_t_70]

**二、SQL：**

`Flink` 提供的最高级别的抽象是 `SQL`。这种抽象在语义和表达方式上均类似于 `Table API`，但是将程序表示为 `SQL` 查询表达式。在 `SQL` 抽象与 `Table API SQL` 查询紧密地相互作用，并且可以在中定义的表执行 `Table API`。

查询

[SQL][]

**三、Table API：**

`Table API 是 Flink` 的语言嵌入式关系 `API`，用于在 `Java 或 Scala` 中编写类 `SQL` 的查询，这些查询会自动进行优化。`Table API` 查询可以使用一致的语法和语义同时在批处理或流数据上运行。

**Table API 支持非常多的 Operations（操作）**

From：与 SQL 查询中的 FROM 子句相似。扫描已注册的表。适用于批处理和流。

Table orders = tableEnv.from("Orders");

Select：类似于 SQL SELECT 语句。执行选择操作。适用于批处理和流。

Table orders = tableEnv.from("Orders");
    Table result = orders.select("a, c as d");

As：重命名字段。适用于批处理和流。

Table orders = tableEnv.from("Orders");
    Table result = orders.as("x, y, z, t");

Where / Filter: 类似于 SQL WHERE 子句。筛选出未通过筛选谓词的行。适用于批处理和流。

Table orders = tableEnv.from("Orders");
    Table result = orders.where("b === 'red'");

AddColumns: 执行字段添加操作。如果添加的字段已经存在，它将引发异常。适用于批处理和流。

Table orders = tableEnv.from("Orders");
    Table result = orders.addColumns("concat(c, 'sunny')");

DropColumns: 执行字段删除操作。字段表达式应该是字段引用表达式，并且只能删除现有字段。适用于批处理和流。

Table orders = tableEnv.from("Orders");
    Table result = orders.dropColumns("b, c");

RenameColumns: 执行字段重命名操作。字段表达式应该是别名表达式，并且只有现有字段可以重命名。适用于批处理和流。

Table orders = tableEnv.from("Orders");
    Table result = orders.renameColumns("b as b2, c as c2");

GroupBy Aggregation: 类似于 SQL GROUP BY 子句。使用以下正在运行的聚合运算符将分组键上的行分组，以逐行聚合行。适用于批处理和流。

Table orders = tableEnv.from("Orders");
    Table result = orders.groupBy("a").select("a, b.sum as d");

Inner Join: 类似于 SQL JOIN 子句。连接两个表。两个表必须具有不同的字段名称，并且至少一个相等的联接谓词必须通过联接运算符或使用 where 或 filter 运算符进行定义。适用于批处理和流。

Table left = tableEnv.fromDataSet(ds1, "a, b, c");
    Table right = tableEnv.fromDataSet(ds2, "d, e, f");
    Table result = left.join(right).where("a = d").select("a, b, e");

Outer Join: 类似于 SQL LEFT / RIGHT / FULL OUTER JOIN 子句。连接两个表。两个表必须具有不同的字段名称，并且必须至少定义一个相等联接谓词。适用于批处理和流。

Table left = tableEnv.fromDataSet(ds1, "a, b, c");
    Table right = tableEnv.fromDataSet(ds2, "d, e, f");
    
    Table leftOuterResult = left.leftOuterJoin(right, "a = d").select("a, b, e");
    Table rightOuterResult = left.rightOuterJoin(right, "a = d").select("a, b, e");
    Table fullOuterResult = left.fullOuterJoin(right, "a = d").select("a, b, e");

**四、DataStream/DataSet API:**

处理有界数据集，对数据集进行转换（例如，过滤，映射，联接，分组）。最初从某些来源（例如，通过读取文件或从本地集合）创建数据集。结果通过接收器返回，接收器可以例如将数据写入（分布式）文件或标准输出（例如命令行终端）。  
`DataStream API 是 Flink` 的主要抽象，用于通过 `Java 或 Scala` 实现具有复杂时间语义的有状态数据流处理的应用程序。

**1、DataSet API（Transformations（转换）、Data Sources（数据源）、Data Sinks（数据接收器））**

`Transformations`（转换）：  
Map：取一个元素并产生一个元素。

data.map(new MapFunction<String, Integer>() { 
      public Integer map(String value) {  return Integer.parseInt(value); }
    });

FlatMap：取一个元素并产生零个，一个或多个元素。

data.flatMap(new FlatMapFunction<String, String>() { 
      public void flatMap(String value, Collector<String> out) { 
        for (String s : value.split(" ")) { 
          out.collect(s);
        }
      }
    });

Filter: 为每个元素评估一个布尔函数，并保留该函数返回 true 的布尔函数。

data.filter(new FilterFunction<Integer>() { 
      public boolean filter(Integer value) {  return value > 1000; }
    });

Reduce: 通过将两个元素重复组合为一个元素，将一组元素组合为一个元素。Reduce 可以应用于完整的数据集或分组的数据集。

data.reduce(new ReduceFunction<Integer> { 
      public Integer reduce(Integer a, Integer b) {  return a + b; }
    });

Aggregate: 将一组值汇总为一个值。可以将聚合功能视为内置的归约功能。汇总可以应用于完整的数据集，也可以应用于分组的数据集。

Dataset<Tuple3<Integer, String, Double>> input = // [...]
    DataSet<Tuple3<Integer, String, Double>> output = input.aggregate(SUM, 0).and(MIN, 2);

**2、DataStream API （Transformations（转换）、Data Sources（数据源）、Data Sinks（数据接收器））**

`Transformations`（转换）  
Map：取一个元素并产生一个元素。下面的例子是通过一个映射函数，将输入流的值加倍。

DataStream<Integer> dataStream = //...
    dataStream.map(new MapFunction<Integer, Integer>() { 
        @Override
        public Integer map(Integer value) throws Exception { 
            return 2 * value;
        }
    });

FlatMap：取一个元素并产生零个，一个或多个元素。FlatMap 功能可将句子拆分为单词：

dataStream.flatMap(new FlatMapFunction<String, String>() { 
        @Override
        public void flatMap(String value, Collector<String> out)
            throws Exception { 
            for(String word: value.split(" ")){ 
                out.collect(word);
            }
        }
    });

Filter: 为每个元素评估一个布尔函数，并保留该函数返回 true 的布尔函数。筛选出零值的筛选器：

dataStream.filter(new FilterFunction<Integer>() { 
        @Override
        public boolean filter(Integer value) throws Exception { 
            return value != 0;
        }
    });

Reduce: 对键控数据流进行 “滚动” 压缩。将当前元素与最后一个减小的值合并，并发出新值。reduce 函数创建部分和流：

keyedStream.reduce(new ReduceFunction<Integer>() { 
        @Override
        public Integer reduce(Integer value1, Integer value2)
        throws Exception { 
            return value1 + value2;
        }
    });

Aggregate: 在键控数据流上滚动聚合。min 和 minBy 之间的区别是 min 返回最小值，而 minBy 返回在此字段中具有最小值的元素（与 max 和 maxBy 相同）。

keyedStream.sum(0);
    keyedStream.sum("key");
    keyedStream.min(0);
    keyedStream.min("key");
    keyedStream.max(0);
    keyedStream.max("key");
    keyedStream.minBy(0);
    keyedStream.minBy("key");
    keyedStream.maxBy(0);
    keyedStream.maxBy("key");

Window：可以在已分区的 KeyedStreams 上定义 Windows。Windows 根据某些特征将每个键中的数据分组（例如，最近 5 秒钟内到达的数据）。

dataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5))); // Last 5 seconds of data

Window Join：在给定键和一个公共窗口上连接两个数据流。

dataStream.join(otherStream)
        .where(<key selector>).equalTo(<key selector>)
        .window(TumblingEventTimeWindows.of(Time.seconds(3)))
        .apply (new JoinFunction () { ...});

Aggregations on windows：聚合窗口的内容。min 和 minBy 之间的区别是 min 返回最小值，而 minBy 返回在此字段中具有最小值的元素（与 max 和 maxBy 相同）。

windowedStream.sum(0);
    windowedStream.sum("key");
    windowedStream.min(0);
    windowedStream.min("key");
    windowedStream.max(0);
    windowedStream.max("key");
    windowedStream.minBy(0);
    windowedStream.minBy("key");
    windowedStream.maxBy(0);
    windowedStream.maxBy("key");

**3、Data Sources（数据源）**

基于文件：

readTextFile(path)/ TextInputFormat- 逐行读取文件，并将其作为字符串返回。
    readTextFileWithValue(path)/ TextValueInputFormat- 逐行读取文件，并将它们作为 StringValues 返回。StringValue 是可变的字符串。
    readCsvFile(path)/ CsvInputFormat- 解析以逗号（或其他字符）分隔的字段的文件。返回元组或 POJO 的数据集。支持基本的 Java 类型及其与 Value 相对应的字段类型。
    readFileOfPrimitives(path, Class)/ PrimitiveInputFormat- 解析以换行符（或其他 char 序列）分隔的原始数据类型的文件，例如 String 或 Integer。
    readFileOfPrimitives(path, delimiter, Class)/ PrimitiveInputFormat- 解析以换行符（或其他 char 序列）分隔的原始数据类型的文件，例如 String 或 Integer 使用给定的分隔符。
    readFile(fileInputFormat, path) - 根据指定的文件输入格式读取（一次）文件。
    readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo)- 这是前两个内部调用的方法。它 path 根据给定的读取文件 fileInputFormat。根据提供的内容 watchType，此源可以定期（每 intervalms）监视路径中的新数据（FileProcessingMode.PROCESS_CONTINUOUSLY），或者处理一次当前路径中的数据并退出（FileProcessingMode.PROCESS_ONCE）。使用 pathFilter，用户可以进一步从文件中排除文件。

基于套接字：

socketTextStream- 从套接字读取。元素可以由定界符分隔。

基于集合：

fromCollection(Collection)- 从 Java.util.Collection 创建数据集。集合中的所有元素必须具有相同的类型。
    fromCollection(Iterator, Class)- 从迭代器创建数据集。该类指定迭代器返回的元素的数据类型。
    fromElements(T ...)- 从给定的对象序列创建数据集。所有对象必须具有相同的类型。
    fromParallelCollection(SplittableIterator, Class)- 从迭代器并行创建数据集。该类指定迭代器返回的元素的数据类型。
    generateSequence(from, to) - 并行生成给定间隔中的数字序列。

通用：

readFile(inputFormat, path)/ FileInputFormat- 接受文件输入格式。
    createInput(inputFormat)/ InputFormat- 接受通用输入格式。

**4、Data Sinks（数据接收器）**

writeAsText()/ TextOutputFormat- 将元素按行写为字符串。通过调用每个元素的 * toString（）* 方法获得字符串。
    writeAsFormattedText()/ TextOutputFormat- 将元素按行写为字符串。通过为每个元素调用用户定义的 * format（）* 方法来获取字符串。
    writeAsCsv(...)/ CsvOutputFormat- 将元组写为逗号分隔的值文件。行和字段定界符是可配置的。每个字段的值来自对象的 * toString（）* 方法。
    print()/ printToErr() - 在标准输出 / 标准错误流上打印每个元素的 toString（）值。可选地，可以提供前缀（msg），该前缀在输出之前。这可以帮助区分不同的打印调用。如果并行度大于 1，则输出之前还将带有产生输出的任务的标识符。
    writeUsingOutputFormat()/ FileOutputFormat- 的方法和自定义文件输出基类。支持自定义对象到字节的转换。
    writeToSocket - 根据一个元素将元素写入套接字 SerializationSchema
    addSink- 调用自定义接收器功能。Flink 捆绑有与其他系统（例如 Apache Kafka）的连接器，这些连接器已实现为接收器功能。
    write()/ FileOutputFormat- 的方法和自定义文件输出基类。支持自定义对象到字节的转换。
    output()/ OutputFormat- 最通用的输出方法，用于不基于文件的数据接收器（例如将结果存储在数据库中）。

**五、Stateful Stream Processing:**

通过 `Process Function 嵌入到 DataStream API` 中。它允许用户自由地处理一个或多个流中的事件，并使用一致的容错状态。此外，用户可以注册事件时间和处理时间回调，从而允许程序实现复杂的计算。

参考：[RangeYan][]  
参考：[API][SQL]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNDQ3MzAx_size_16_color_FFFFFF_t_70]: /images/20230521/5237e1b8059844b994e58fa140f93bb6.png
[SQL]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/sql/queries.html
[RangeYan]: http://www.imooc.com/read/64/article/1442