Flink基础之API，DataSet、DataStream、批、流

电玩女神 2021-10-19 03:46 353阅读 0赞

**目录**

DataSet 和DataStream

Anatomy of a Flink Program（剖析Flink程序）

Lazy Evaluation（延迟计算）

Specifying Keys（指定的键）

Specifying Transformation Functions（指定转换功能）

Supported Data Types（支持的数据类型）

Java Tuples and Scala Case Classes

Java POJOs

Primitive Types（基本数据类型）

Regular Classes

Values

Hadoop Writables

Special Types

java类型推断

Accumulators & Counters（累加器和计数）

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

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

根据数据源的类型，即有界或无界数据源，可以编写一个批处理程序或流处理程序，其中DataSet API用于批处理，DataStream API用于流处理。本文将介绍这两种API所共有的基本概念。

**注释：**在展示如何使用这些API的实际示例时，我们将使用StreamingExecutionEnvironment和DataStream API。DataSet API中的概念完全相同，只是替换为ExecutionEnvironment和DataSet。

## DataSet 和DataStream ##

Flink有特殊的类DataSet和DataStream来表示程序中的数据。您可以将它们看作是不可变的数据集合，可以包含重复的数据。在DataSet的情况下，数据是有限的，而对于DataStream，元素的数量可以是无界的。

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

集合最初是通过在Flink程序中添加一个源创建的，通过使用map、filter等API方法对这些源进行转换，可以派生出新的集合。

## Anatomy of a Flink Program（剖析Flink程序） ##

Flink程序看起来像转换数据集合的常规程序。每个程序都由相同的基本部分组成：

1.  初始化执行环境；
2.  加载/创建初始数据；
3.  指定对该数据的转换；
4.  指定计算结果的存储位置；
5.  执行程序

StreamExecutionEnvironment是所有Flink程序的基础。你可以使用这些静态方法在StreamExecutionEnvironment上获得：

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

通常，只需要使用getExecutionEnvironment()，因为这将根据上下文：做正确的事如果你执行程序在IDE或普通Java程序将创建一个本地环境,将执行程序在本地机器上。如果您从程序中创建了一个JAR文件，并通过命令行调用它，Flink集群管理器将执行您的主方法，getExecutionEnvironment()将返回一个执行环境，用于在集群上执行您的程序。

为了指定数据源，执行环境有几个方法可以从文件中读取数据：您可以将它们作为CSV文件逐行读取，或者使用完全定制的数据输入格式读取。要将文本文件作为行序列读取，可以使用：

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

这将为您提供一个数据流，您可以在该数据流上应用转换来创建新的派生数据流。

通过使用转换函数调用DataStream上的方法来应用转换。例如：

DataStream<String> input = ...;
    
    DataStream<Integer> parsed = input.map(new MapFunction<String, Integer>() {
        @Override
        public Integer map(String value) {
            return Integer.parseInt(value);
        }
    });

这将通过将原始集合中的每个字符串转换为整数来创建一个新的DataStream。

一旦有了包含最终结果的DataStream，就可以通过创建接收器将其写入外部系统。下面是一些创建接收器的例子：

writeAsText(String path)
    
    print()

一旦指定了完整的程序，就需要通过调用StreamExecutionEnvironment上的execute()来触发程序执行。根据ExecutionEnvironment的类型，执行将在本地机器上触发，或者将程序提交到集群上执行。

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

## Lazy Evaluation（延迟计算） ##

所有Flink程序都是延迟执行的:当程序的主方法执行时，数据加载和转换不会直接发生。相反，每个操作都被创建并添加到程序的计划中。当执行环境上的execute()调用显式触发执行时，实际执行操作。程序是在本地执行还是在集群上执行取决于执行环境的类型。

延迟计算允许构建复杂的程序，Flink将这些程序作为一个整体规划的单元执行。

## Specifying Keys（指定的键） ##

一些转换(join、coGroup、keyBy、groupBy)要求在元素集合上定义一个键。其他转换(Reduce、GroupReduce、Aggregate、Windows)允许在应用数据之前根据键对数据进行分组。

DataSet分组为：

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

key可以再DataStream上使用

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

Flink的数据模型不是基于键值对的。因此不需要将数据集类型物理地打包到键和值中。键是“虚拟的”，它们被定义为实际数据之上的函数，用于指导分组操作符。

下面将使用DataStream API和keyBy。对于DataSet API，只需用DataSet和groupBy替换即可。

**Define keys for Tuples（使用元组定义键）**

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

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作为键(整数和浮点数作为键)。如果您想要“导航”到嵌套的Tuple2，您必须使用下面解释的字段表达式键。

**Define keys using Field Expressions（使用字段表达式定义键）**

您可以使用基于字符串的字段表达式引用嵌套字段，并定义用于分组、排序、连接或协分组的键。

字段表达式使得在(嵌套的)组合类型(如Tuple和POJO类型)中选择字段非常容易。

在下面的示例中，我们有一个WC POJO，其中包含两个字段“word”和“count”。要按字段词进行分组，只需将其名称传递给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*/);

字段表达式语法：

1.  根据字段名选择POJO字段。例如，“user”引用POJO类型的“user”字段。
2.  根据元组字段名或0偏移字段索引选择元组字段。例如，“f0”和“5”分别引用Java元组类型的第一个和第六个字段。
3.  您可以选择pojo和元组中的嵌套字段。例如“用户。zip是指存储在POJO类型的“user”字段中的POJO的“zip”字段。支持任意嵌套和混合pojo和元组，比如“f1.user”。zip”或“user.f3.1.zip”。
4.  您可以使用“\*”通配符表达式选择完整类型。这也适用于非元组或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;
    }

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

1.  "count"：WC类中的count字段。
2.  "complex"：递归地选择POJO类型ComplexNestedClass的字段复合体的所有字段。
3.  "complex.word.f2"：选择嵌套Tuple3的最后一个字段。
4.  "complex.hadoopCitizen"：选择IntWritable类型

**Define keys using Key Selector Functions（使用键选择器函数定义键）**

定义键的另一种方法是“键选择器”函数。键选择器函数接受单个元素作为输入，并返回该元素的键。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; }
       });

### Specifying Transformation Functions（指定转换功能） ###

大多数转换都需要用户定义的函数。

**Java**

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

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); }
    });

Flink还支持Java API中的Java 8 Lambdas：

data.filter(s -> s.startsWith("http://"));
    data.reduce((i1,i2) -> i1 + i2);

所有需要用户定义函数的转换都可以将一个富函数作为参数。例如：

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); }
    };

并像往常一样将函数传递给map变换：

data.map(new MyMapFunction());

富函数也可以定义为一个匿名类：

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

除了用户定义的函数(map、reduce等)之外，富函数还提供四种方法：`open`, `close`, `getRuntimeContext`, 和 `setRuntimeContext`。这对于参数化函数(请参阅[Passing Parameters to Functions][])、创建和结束本地状态、访问广播变量(请参阅[Broadcast Variables][]))以及访问运行时信息，如累加器和计数器(请参阅[Accumulators and Counters][])，以及关于迭代的信息(请参阅[Iterations][])，都非常有用。

**Scala**

正如在前面的例子中已经看到的，所有的操作都接受lambda函数来描述操作：

val data: DataSet[String] = // [...]
    data.filter { _.startsWith("http://") }
    
    val data: DataSet[Int] = // [...]
    data.reduce { (i1,i2) => i1 + i2 }
    // or
    data.reduce { _ + _ }

所有以lambda函数为参数的转换都可以以富函数为参数。例如：

data.map { x => x.toInt }

可以写：

class MyMapFunction extends RichMapFunction[String, Int] {
      def map(in: String):Int = { in.toInt }
    };

并将函数传递给map变换：

data.map(new MyMapFunction())

富函数也可以定义为一个匿名类：

data.map (new RichMapFunction[String, Int] {
      def map(in: String):Int = { in.toInt }
    })

## Supported Data Types（支持的数据类型） ##

Flink对数据集或数据流中的元素类型进行了一些限制。原因是系统分析类型以确定有效的执行策略。

以下6种不同的数据类型：

*  ### **Java Tuples** and **Scala Case Classes** ###

Java：提供了Tuple1到Tuple25，元组的每个字段都可以是任意的Flink类型，包括further tuples，从而产生嵌套的元组。可以使用字段名作为元组直接访问元组的字段。或者使用通用getter()，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")

Scala：Scala tuples是一个特殊的Case类，是包含不同类型的固定数量字段的组合类型，Tuple fields根据首字段的\_1偏移量来寻址，Case类字段按其名称访问。

case class WordCount(word: String, count: Int)
    val input = env.fromElements(
        WordCount("hello", 1),
        WordCount("world", 2)) // Case Class Data Set
    
    input.keyBy("word")// key by field expression "word"
    
    val input2 = env.fromElements(("hello", 1), ("world", 2)) // Tuple2 Data Set
    
    input2.keyBy(0, 1) // key by field positions 0 and 1

*  ### **Java POJOs** ###

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

1.  必须是公共类
2.  有默认构造函数
3.  所有字段要么是公共的，要么必须通过getter和setter函数访问。对于一个名为foo的字段，getter和setter方法必须分别命名为getFoo()和setFoo()。
4.  字段的类型必须由Flink支持。目前，Flink使用Avro序列化任意对象(比如Date)。

Flink分析POJO类型的结构，即，它学习POJO的领域。因此，POJO类型比一般类型更容易使用。此外，Flink可以比一般类型更有效地处理pojo。

// Java
    
    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"
    
    // Scala
    
    
    class WordWithCount(var word: String, var count: Int) {
        def this() {
          this(null, -1)
        }
    }
    
    val input = env.fromElements(
        new WordWithCount("hello", 1),
        new WordWithCount("world", 2)) // Case Class Data Set
    
    input.keyBy("word")// key by field expression "word"

*  ### **Primitive Types（基本数据类型）** ###

Flink支持所有Java和Scala基本类型。

*  ### **Regular Classes** ###

Flink支持大多数Java和Scala类(API和定制)。限制适用于包含不能序列化字段的类，如文件指针、I/O流或其他本机资源。通常，遵循Java bean约定的类都支持的很好。

所有未标识为POJO类型的类(请参阅上面的POJO需求)都由Flink作为通用类类型处理。Flink将这些数据类型视为黑盒，无法访问它们的内容(例如，提高排序效率)。一般类型使用序列化框架Kryo反序列化。

*  ### **Values** ###

值类型自定义它们的序列化和反序列化。它们没有使用通用的序列化框架，而是通过实现`org.apache.flinktypes.Value`为这些操作提供定制代码read()和write()。当使用通用序列化效率非常低时，使用Values Types是合理的。例如，数据类型将实现元素的稀疏向量作为数组。知道数组大部分为零，就可以对非零元素使用特殊编码，而通用串行化只需编写所有数组元素。

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

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

*  ### **Hadoop Writables** ###

实现`org.apache.hadoop.Writable接口，在write()和readFields()方法中自定义序列化逻辑。`

*  ### **Special Types** ###

`特殊的类型，包括Scala的Either`, `Option`, and `Try，Java也有类似Either的自定义实现，用于处理错误，输出两种不同类型的操作记录。`

*  ### **`java类型推断`** ###

Java编译器在编译之后会丢弃很多泛型类型信息。这在Java中称为类型擦除。这意味着在运行时，对象的实例不再知道它的泛型类型。例如，DataStream<String>和DataStream<Long>的实例在JVM中看起来是一样的。

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

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

可以通过输入格式和函数实现[ResultTypeQueryable][]接口，从而显式地告诉API它们的返回类型。函数调用的输入类型通常可以由前面操作的结果类型推断。

## Accumulators & Counters（累加器和计数） ##

累加器是带有一个**add操作**和一个**最终的累积结果**的简单构造，该结果在作业结束后可用。

最直接的累加器是**计数器**：可以使用`Accumulator.add(V value)`的方法递增它。作业结束时，Flink将汇总(合并)所有部分结果，并将结果发送给客户机。累加器在调试期间非常有用，如果想快速了解更多有关数据的信息，也可以使用累加器。

Flink目前有以下内置累加器，每一个都实现了Accumulator接口：

*  IntCounter, LongCounter and DoubleCounter：参见下面计数器的实例
 *  Histogram（直方图）：针对离散数量的箱子的直方图实现。在内部，它只是一个从整数到整数的映射。您可以使用它来计算值的分布，例如单词计数程序的每行单词的分布。

**如何使用：**

第一步，需要创建一个accumulator对象。

private IntCounter numLines = new IntCounter();

第二步，注册accumulator对象，通常实在rich函数的open()方法中，如下

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

现在你可以在operator function的任何地方使用这个对象，包括open()和close()方法中。

this.numLines.add(1);

整个结果将存储在JobExecutionResult对象中，该对象是从执行环境的execute()方法返回的(目前，只有在执行等待作业完成时才有效)。

myJobExecutionResult.getAccumulatorResult("num-lines")

所有累加器对每个作业共享一个名称空间。因此，可以在作业的不同操作符函数中使用相同的累加器。Flink将在内部合并所有具有相同名称的累加器。

关于累加器和迭代的说明：当前，累加器的结果只有在整个作业结束后才可用。Flink计划让上一个迭代的结果在下一个迭代中可用。可以使用聚合器来计算每次迭代的统计数据，迭代的终止基于这样的统计数据。

**自定义累加器：**

要实现自己的累加器，只需编写累加器接口的实现即可。如果认为自定义累加器应该与Flink一起提供，请随意创建一个pull请求。

可以选择实现[Accumulator][] or [SimpleAccumulator][]。

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

[Passing Parameters to Functions]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/batch/index.html#passing-parameters-to-functions
[Broadcast Variables]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/batch/index.html#broadcast-variables
[Accumulators and Counters]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/api_concepts.html#accumulators--counters
[Iterations]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/batch/iterations.html
[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
[SimpleAccumulator]: https://github.com/apache/flink/blob/master//flink-core/src/main/java/org/apache/flink/api/common/accumulators/SimpleAccumulator.java