【零基础学flink】flink中常用API详解

快来打我* 2022-02-14 07:51 616阅读 0赞

Flink程序是实现分布式集合转换的常规程序（例如， filtering, mapping, updating state, joining, grouping, defining windows, aggregating）。最初从源创建集合（例如，通过从文件，kafka主题或从本地的内存集合中读取）。结果通过接收器返回，接收器可以例如将数据写入（分布式）文件或标准输出（例如，命令行终端）。 Flink程序可以在各种环境中运行，独立运行（standalone）或嵌入其他程序中。

执行可以在本地JVM中执行，也可以在许多计算机的集群上执行。 根据数据源的类型，即有界或无界源，您可以编写批处理程序或流程序，其中DataSet API用于批处理，DataStream API用于流式处理。本指南将介绍两种API共有的基本概念，但请参阅我们的流媒体指南（ [Streaming Guide][]）和批处理指南（ [Batch Guide][]），了解有关使用每个API编写程序的具体信息。

注：在演示如何使用API的实际示例时，我们将使用StreamingExecutionEnvironment和DataStream API。 DataSet API中的概念完全相同，只需用ExecutionEnvironment和DataSet替换即可。

## **目录** ##

*  DataSet和DataStream
 *  Flink计划的剖析
 *  懒惰的评价
 *  指定密钥
 *  指定转换函数
 *  支持的数据类型
 *  蓄电池和计数器

## DataSet和DataStream ##

Flink具有特殊类`DataSet`并`DataStream`在程序中表示数据。您可以将它们视为可以包含重复项的不可变数据集合。`DataSet`中的数据是有限（bound）的情况，对于一个`DataStream`元素数量是无界（unbound）的。

这些集合在某些方面与常规Java集合不同。首先，它们是不可变的，这意味着一旦创建它们就无法添加或删除元素。你也不能简单地检查里面的元素。

集合最初通过在flink程序添加源创建的，对原始集合的transformation操作会产生新的集合，常见的transformation操作API：`map`，`filter`等等。

## Flink程序的剖析 ##

Flink程序看起来像是转换数据集合的常规程序。每个程序都包含以下的几个部分：

1.  获得一个`execution environment`，
2.  加载/创建初始数据（数据源source），
3.  指定此数据的转换（transformation），
4.  指定放置计算结果的位置，
5.  触发程序执行

*  [**Java的**][Java]
 *  [**斯卡拉**][Link 1]

我们现在将概述每个步骤，更多详细信息请参阅相应部分以获取。请注意，Java DataSet API的所有核心类都可以在[org.apache.flink.api.java][]包中找到， 而Java DataStream API的类可以在[org.apache.flink.streaming.api中][org.apache.flink.streaming.api]找到 。

这`StreamExecutionEnvironment`是所有Flink程序的基础。您可以使用以下静态方法获取一个`StreamExecutionEnvironment`：

getExecutionEnvironment()
    
    createLocalEnvironment()
    
    createRemoteEnvironment(String host, int port, String... jarFiles)

通常，您只需要使用`getExecutionEnvironment()`，因为这将根据上下文做正确的事情：如果您在IDE中执行程序或作为常规Java程序，它将创建一个本地环境，将在本地计算机上执行您的程序。如果您从程序中创建了一个JAR文件，并通过[命令行][Link 2]调用它 ，则Flink集群管理器将执行您的main方法并`getExecutionEnvironment()`返回一个执行环境，以便在集群上执行您的程序。

对于指定数据源，执行环境有几种方法可以使用各种方法从文件中读取：您可以逐行读取它们，CSV文件或使用完全自定义数据输入格式。要将文本文件作为一系列行读取，您可以使用：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
    DataStream<String> text = env.readTextFile("file:///path/to/file");

这将为您提供一个DataStream，然后您可以在其上应用转换来创建新的派生DataStream。

您可以通过使用转换函数调用DataStream上的方法来应用转换。例如，地图转换如下所示：

DataStream<String> input = ...;
    // <String, Integer>中的String表示的是stream中的数据类型，Integer表示的是map输出的类型
    DataStream<Integer> parsed = input.map(new MapFunction<String, Integer>() {
        @Override
        public Integer map(String value) {
            return Integer.parseInt(value);
        }
    });

这将创建新一个新的DataStream：通过将原始集合中的每个String转换为Integer。

一旦有了包含最终结果的DataStream（一系列转换操作之后），就可以通过创建接收器（sink）将其写入外部系统。下面是创建接收器的一些示例方法：

writeAsText(String path)
    
    print()

在写完所有的代码逻辑之后，你需要**触发执行程序**调用`StreamExecutionEnvironment`的 `execute()`方法。根据`ExecutionEnvironment`执行类型，将在本地计算机上触发执行或提交程序以在群集上执行。

该`execute()`方法返回一个`JobExecutionResult`，它包含执行时间和累加器结果。

有关流数据源和接收器的信息，请参阅[流指南][Streaming Guide]，以及有关DataStream上支持的转换的更深入信息。

有关批处理数据源和接收器的信息，请查看[批处理指南][Batch Guide]，以及有关DataSet支持的转换的更深入信息。

## Lazy execution（懒加载） ##

所有的Flink程序都是懒惰（lazy）地执行：当执行程序的main方法时，数据加载和转换不会直接发生。而是创建每个操作并将其添加到程序的计划中。当`execute()`方法被显示调用，才触发实际执行操作。程序是在本地执行还是在集群上执行取决于执行环境的类型

## 指定key ##

某些转换（join，coGroup，keyBy，groupBy）要求在元素集合上定义键（key）。其他转换（Reduce，GroupReduce，Aggregate，Windows）允许数据在应用之前在键（key）上分组。

DataSet分组为

DataSet<...> input = // [...]
    DataSet<...> reduced = input
      .groupBy(/*define key here*/)
      .reduceGroup(/*do something*/);

可以使用下面方式在DataStream上指定键（key）

DataStream<...> input = // [...]
    DataStream<...> windowed = input
      .keyBy(/*define key here*/)
      .window(/*window specification*/);

Flink的数据模型**不是基于键值对**。因此，您无需将数据集类型物理打包到键和值中。在flink中，**键是“虚拟的”**：它们被定义为实际数据上的函数，以指导分组算子。

\*\*注意：\*\*在下面的讨论中，我们将使用`DataStream`API和`keyBy`。对于DataSet API，您只需要替换为`DataSet`和`groupBy`。

### 定义元组的键 ###

最简单的情况是在元组的一个或多个字段上对元组进行分组：

DataStream<Tuple3<Integer,String,Long>> input = // [...]
    KeyedStream<Tuple3<Integer,String,Long>,Tuple> keyed = input.keyBy(0)

元组在**第一个字段**（整数类型）上分组。

DataStream<Tuple3<Integer,String,Long>> input = // [...]
    KeyedStream<Tuple3<Integer,String,Long>,Tuple> keyed = input.keyBy(0,1)

在这里，我们将元组分组在由**第一个和第二个字段**组成的复合键上。

关于嵌套元组的注释：如果你有一个带有嵌套元组的DataStream，例如：

DataStream<Tuple3<Tuple2<Integer, Float>,String,Long>> ds;

指定`keyBy(0)`将导致系统使用 `Tuple2`作为键（以Integer和Float为键）。如果要访问嵌套中`Tuple2`，则必须使用下面域表达式

### 使用域表达式（Field Expressions定义键） ###

您可以使用基于字符串的字段表达式来引用嵌套字段，并定义用于grouping，sorting，joining或coGrouping的键。

字段表达式可以非常轻松地选择（嵌套）复合类型中的字段，例如Tuple和POJO类型。

在下面的示例中，我们有一个`WC`POJO，其中包含两个字段“word”和“count”。要按字段分组`word`，我们只需将其名称传递给`keyBy()`函数即可。

// some ordinary POJO (Plain old Java Object)
    public class WC {
      public String word;
      public int count;
    }
    DataStream<WC> words = // [...]
    DataStream<WC> wordCounts = words.keyBy("word").window(/*window specification*/);

**字段表达式语法**：

*  按字段名称选择POJO字段。例如，`"user"`指POJO类型的“user”字段。
 *  按字段名称或0偏移字段索引选择元组字段。例如`"f0"`，分别`"5"`引用Java Tuple类型的`第一个`和`第六个`字段。
 *  您可以在POJO和Tuples中选择嵌套字段。例如，`"user.zip"`指POJO的“zip”字段，其存储在POJO类型的“user”字段中。支持POJO和元组的任意嵌套和混合，例如`"f1.user.zip"`或`"user.f3.1.zip"`。
 *  您可以使用`"*"`通配符表达式选择完整类型。这也适用于非Tuple或POJO类型的类型。

**字段表达示例**：

public static class WC {
      public ComplexNestedClass complex; //nested POJO
      private int count;
      // getter / setter for private field (count)
      public int getCount() {
        return count;
      }
      public void setCount(int c) {
        this.count = c;
      }
    }
    public static class ComplexNestedClass {
      public Integer someNumber;
      public float someFloat;
      public Tuple3<Long, Long, String> word;
      public IntWritable hadoopCitizen;
    }

这些是上面示例代码的有效字段表达式：

*  `"count"`：类中的count字段`WC`。
 *  `"complex"`：类中的POJO类型的字段，复合体的所有字段`ComplexNestedClass`。
 *  `"complex.word.f2"`：选择嵌套的最后一个字段`Tuple3`。
 *  `"complex.hadoopCitizen"`：选择Hadoop `IntWritable`类型。

### 使用键选择器（keySelector()）功能定义key ###

定义key的另一种方法是“键选择器（keySelector）”功能。键选择器函数将单个元素作为输入并返回元素的键。key可以是任何类型，并且可以从确定性计算中导出。

以下示例显示了一个键选择器函数，它只返回一个对象的字段：

// some ordinary POJO
    public class WC {public String word; public int count;}
    DataStream<WC> words = // [...]
    KeyedStream<WC> keyed = words
      .keyBy(new KeySelector<WC, String>() {
         public String getKey(WC wc) { return wc.word; }
       });

## 指定转换函数 ##

大多数转换都需要用户定义的函数。本节列出了定义转换函数的不同方法

#### 实现接口 ####

最基本的方法是实现一个提供的接口：

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

#### 匿名函数 ####

你可以将函数作为匿名类传递：

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

#### Java 8 Lambdas ####

Flink还支持Java API中的Java 8 Lambdas表达式。

data.filter(s -> s.startsWith("http://"));

data.reduce((i1,i2) -> i1 + i2);

#### rich functions ####

在上面，我们需要定义转换函数，实际上通过*rich function*也可以实现。例如：

class MyMapFunction implements MapFunction<String, Integer> {
      public Integer map(String value) { return Integer.parseInt(value); }
    };

下面与上面等价，上面也可以写成：

class MyMapFunction extends RichMapFunction<String, Integer> {
      public Integer map(String value) { return Integer.parseInt(value); }
    };

并像往常一样将MyMapFunction 函数传递给`map`转换函数：

data.map(new MyMapFunction());

rich函数也可以定义为匿名类：

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

rich函数，除了提供用户定义的函数（map，reduce等），还有另外四种方法：`open`，`close`，`getRuntimeContext`，和 `setRuntimeContext`。这些是非常有用的参数化函数（请参阅[将参数传递给函数][Link 3]），创建和完成本地状态，访问广播变量（请参阅 [广播变量][Link 4]）以及访问运行时信息（如累加器和计数器）（请参阅 [累加器和计数器][Link 5]）以及有关信息的信息。迭代（参见[迭代][Link 6]）。

## 支持的数据类型 ##

Flink对DataSet或DataStream中可以包含的元素类型设置了一些限制。

主要有六种不同类别的数据类型：

*  Java Tuples and Scala Case Classes
 *  Java POJOs
 *  Primitive Types
 *  Regular Classes
 *  Values
 *  Hadoop Writables
 *  Special Types

#### 元组和case class ####

元组是包含固定数量的具有各种类型的字段的复合类型。Java API提供了`Tuple1`到`Tuple25` class。元组的每个字段都可以是包含更多元组（嵌套）的任意Flink类型，从而产生嵌套元组。可以使用字段名称直接访问元组的字段`tuple.f4`，或使用通用getter方法 `tuple.getField(int position)`。字段索引从0开始。请注意，这与Scala元组形成对比，但它与Java的索引更为一致。

DataStream<Tuple2<String, Integer>> wordCounts = env.fromElements(
        new Tuple2<String, Integer>("hello", 1),
        new Tuple2<String, Integer>("world", 2));
    
    wordCounts.map(new MapFunction<Tuple2<String, Integer>, Integer>() { 
        @Override
        public Integer map(Tuple2<String, Integer> value) throws Exception { 
            return value.f1;
        }
    });
    
    wordCounts.keyBy(0); // also valid .keyBy("f0")

#### POJO ####

如果满足以下要求，则Flink将Java和Scala类视为特殊的POJO数据类型：

*  public 修饰的class
 *  它必须有一个没有参数的公共构造函数（默认构造函数）。
 *  所有字段都是公共的，或者必须可以通过getter和setter函数访问。对于一个名为`foo`的属性，它的getter和setter方法的字段必须命名`getFoo()`和`setFoo()`。
 *  属性的类型必须是Flink支持字段的类型。目前，Flink使用[Avro][]序列化任意对象（例如`Date`）。

Flink分析了POJO类型的结构，即它了解了POJO的字段。因此，POJO类型比一般类型更容易使用。此外，Flink可以比一般类型更有效地处理POJO。

以下示例显示了一个包含两个公共字段的简单POJO。

public class WordWithCount {
    
        public String word;
        public int count;
    
        public WordWithCount() {}
    
        public WordWithCount(String word, int count) {
            this.word = word;
            this.count = count;
        }
    }
    
    DataStream<WordWithCount> wordCounts = env.fromElements(
        new WordWithCount("hello", 1),
        new WordWithCount("world", 2));
    
    wordCounts.keyBy("word"); // key by field expression "word"

#### 原始类型（Primitive Types） ####

flink支持所有Java和Scala的原始类型，如`Integer`，`String`和`Double`。

#### 一般类别（General Class Types） ####

Flink支持大多数Java和Scala类（API和自定义）。限制适用于包含无法序列化的字段的类，如文件指针，I / O流或其他本机资源。遵循Java Beans约定的类通常可以很好地工作。

所有未标识为POJO类型的类（请参阅上面的POJO要求）都由Flink作为常规类类型处理。Flink将这些数据类型视为黑盒子，并且无法访问其内容（即，用于有效排序）。使用序列化框架[Kryo][]对常规类型进行反序列化。

#### 值（Values） ####

*值*类型手动描述其序列化和反序列化。它们不是通过通用序列化框架，而是通过`org.apache.flinktypes.Value`使用方法`read`和实现接口为这些操作提供自定义代码`write`。当通用序列化效率非常低时，使用值类型是合理的。一个示例稀疏数组，数组大部分为零，可以对非零元素使用特殊编码，而通用序列化需编写所有数组元素。

该`org.apache.flinktypes.CopyableValue`接口以类似的方式支持手动内部克隆逻辑。

Flink带有与基本数据类型对应的预定义值类型。（`ByteValue`， `ShortValue`，`IntValue`，`LongValue`，`FloatValue`，`DoubleValue`，`StringValue`，`CharValue`， `BooleanValue`）。这些Value类型充当基本数据类型的可变变体：它们的值可以更改，允许程序员重用对象，减轻垃圾收集器中的压力。

#### Hadoop Writables ####

您可以使用实现该`org.apache.hadoop.Writable`接口的类型。`write()`和`readFields()`方法中定义的序列化逻辑将用于序列化。

#### 特殊类型 ####

您可以使用特殊类型，包括Scala的`Either`，`Option`和`Try`。Java API有自己的自定义实现`Either`。与Scala类似`Either`，它代表两种可能类型的值，*左*或*右*。 `Either`可用于错误处理或需要输出两种不同类型记录的运算符。

#### 类型擦除和类型推断 ####

*注意：本节仅适用于Java。*

Java编译器在编译后抛弃了大部分泛型类型信息。这在Java中称为`*类型擦除*`。这意味着在运行时，对象的实例不再知道其泛型类型。例如，对JVM来说`DataStream<String>`和`DataStream<Long>`是一样的。

Flink在准备执行程序时（当调用程序的主要方法时）需要类型信息。Flink Java API尝试重建以各种方式丢弃的类型信息，并将其显式存储在数据集和运算算子中。您可以通过检索类型`DataStream.getType()`。该方法返回一个实例`TypeInformation`，这是Flink表示类型的内部方式。

类型推断有其局限性，在某些情况下需要程序员的“合作”。这方面的示例是从集合创建数据集的方法，例如`ExecutionEnvironment.fromCollection(),`您可以传递描述类型的参数的位置。但是通用函数`MapFunction<I, O>`也可能需要额外的类型信息。

[ResultTypeQueryable][] 接口可以通过输入格式和功能来实现明确地告诉API他们的返回类型。调用函数的*输入类型*通常可以通过先前操作的结果类型来推断。

## 累加器和计数器 ##

累加器是具有**add操作**和**最终结果**简单结构，可在作业结束后使用。

最简单的累加器是**计数器**：您可以使用该`Accumulator.add(V value)`方法递增它 。在工作结束时，Flink将汇总（合并）所有部分结果并将结果发送给客户。在调试过程中，或者如果您想快速了解有关数据的更多信息，累加器非常有用。

Flink目前有以下**内置累加器**。它们中的每一个都实现了 [Accumulator][] 接口。

*  [**IntCounter**][IntCounter]， [**LongCounter**][LongCounter] 和[\*\* DoubleCounter\*\*][DoubleCounter]：请参阅下面的使用计数器的示例。
 *  [**直方图**][Link 7]：离散数量的区间的直方图实现。在内部，它只是一个从Integer到Integer的映射。您可以使用它来计算值的分布，例如字数统计程序的每行字数的分布。

**如何使用累加器：**

首先，您必须在要创建累加器对象（此处为计数器）。

private IntCounter numLines = new IntCounter();

其次，您必须注册累加器对象，通常在*rich*函数的`open()`方法中 。在这里您还可以定义累加器的名称。

getRuntimeContext().addAccumulator("num-lines", this.numLines);

您现在可以在运算算子中的任何位置使用累加器，包括在`open()`和 `close()`方法中。

this.numLines.add(1);

整个结果将存储在`JobExecutionResult`从`execute()`执行环境的方法返回的对象中（当前这仅在执行等待作业完成时才有效）。

myJobExecutionResult.getAccumulatorResult("num-lines")

所有累加器每个作业共享一个命名空间。因此，您可以在作业的不同的转换算子中使用相同的累加器。Flink将在内部合并所有具有相同名称的累加器。

关于累加器和迭代的注释：目前累加器的结果仅在整个作业结束后才可用。我们还计划在下一次迭代中使前一次迭代的结果可用。您可以使用 [聚合器][Link 8] 来计算每次迭代统计信息，并根据此类统计信息确定迭代的终止。

**定制累加器：**

要实现自己的累加器，只需编写Accumulator接口的实现即可。如果您认为您的自定义累加器应与Flink一起提供，请随意在github上拉取请求。

您可以选择实施 [Accumulator][] 或[SimpleAccumulator][]。

`Accumulator<V,R>`最灵活：它定义`V`要添加的值的类型，`R`最终结果的结果类型。例如，对于直方图，`V`是数字并且`R`是直方图。`SimpleAccumulator`适用于两种类型相同的情况，例如柜台。

--------------------

扫描下方二维码，**及时**获取更多**互联网求职面经**、**java**、**python**、**爬虫**、**大数据**等技术，和**海量资料分享**：  
公众号\*\*`菜鸟名企梦`**后台发送“csdn”即可免费领取【csdn】和【百度文库】下载服务；  
公众号**`菜鸟名企梦`**后台发送“资料”:即可领取**5T精品学习资料\*\*、**java面试考点**和**java面经总结**，以及**几十个java、大数据项目**，**资料很全，你想找的几乎都有**  
![扫码关注，及时获取更多精彩内容。（博主今日头条大数据工程师）][1240]

[Streaming Guide]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/datastream_api.html
[Batch Guide]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/batch/index.html
[Java]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/api_concepts.html#tab_java_0
[Link 1]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/api_concepts.html#tab_scala_0
[org.apache.flink.api.java]: https://github.com/apache/flink/blob/master//flink-java/src/main/java/org/apache/flink/api/java
[org.apache.flink.streaming.api]: https://github.com/apache/flink/blob/master//flink-streaming-java/src/main/java/org/apache/flink/streaming/api
[Link 2]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/cli.html
[Link 3]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/batch/index.html#passing-parameters-to-functions
[Link 4]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/batch/index.html#broadcast-variables
[Link 5]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/api_concepts.html#accumulators--counters
[Link 6]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/batch/iterations.html
[Avro]: http://avro.apache.org/
[Kryo]: https://github.com/EsotericSoftware/kryo
[ResultTypeQueryable]: https://github.com/apache/flink/blob/master//flink-core/src/main/java/org/apache/flink/api/java/typeutils/ResultTypeQueryable.java
[Accumulator]: https://github.com/apache/flink/blob/master//flink-core/src/main/java/org/apache/flink/api/common/accumulators/Accumulator.java
[IntCounter]: https://github.com/apache/flink/blob/master//flink-core/src/main/java/org/apache/flink/api/common/accumulators/IntCounter.java
[LongCounter]: https://github.com/apache/flink/blob/master//flink-core/src/main/java/org/apache/flink/api/common/accumulators/LongCounter.java
[DoubleCounter]: https://github.com/apache/flink/blob/master//flink-core/src/main/java/org/apache/flink/api/common/accumulators/DoubleCounter.java
[Link 7]: https://github.com/apache/flink/blob/master//flink-core/src/main/java/org/apache/flink/api/common/accumulators/Histogram.java
[Link 8]: https://github.com/apache/flink/blob/master//flink-java/src/main/java/org/apache/flink/api/java/operators/IterativeDataSet.java#L98
[SimpleAccumulator]: https://github.com/apache/flink/blob/master//flink-core/src/main/java/org/apache/flink/api/common/accumulators/SimpleAccumulator.java
[1240]: /images/20220214/145b417e7f77465589cff6aedc52da7a.png