Flink实战(四) Flink 基本概念

港控/mmm° 2023-10-11 10:04 8阅读 0赞

### 文章目录 ###

*   *  1. DataSet and DataStream
     *  2. 剖析 Flink程序
     *  3. 延迟计算
     *  4. key 的 定义方式
     *   *  定义元组的key
         *  使用字段表达式定义key
         *  使用KeySelector 定义Key
     *  5. 转换函数
     *   *   *  Implementing an interface
             *  Anonymous classes
             *  Java 8 Lambdas
             *  Rich functions
     *  6. 支持的数据类型
     *  7 累加器&计数器

> Flink programs are regular programs that implement transformations on distributed collections (e.g., filtering, mapping, updating state, joining, grouping, defining windows, aggregating).
> 
> Collections are initially created from sources (e.g., by reading from files, kafka topics, or from local, in-memory collections).
> 
> Results are returned via sinks, which may for example write the data to (distributed) files, or to standard output (for example, the command line terminal).
> 
> Flink programs run in a variety of contexts, standalone, or embedded in other programs. The execution can happen in a local JVM, or on clusters of many machines.

## 1. DataSet and DataStream ##

Flink has the special classes `DataSet` and `DataStream` to represent data in a program. You can think of them as immutable collections of data that can contain duplicates. In the case of `DataSet` the data is finite while for a `DataStream` the number of elements can be unbounded.

These collections differ from regular Java collections in some key ways. First, they are immutable, meaning that once they are created you cannot add or remove elements. You can also not simply inspect the elements inside.

A collection is initially created by adding a source in a Flink program and new collections are derived from these by transforming them using API methods such as `map`, `filter` and so on.

`Flink` 中最重要的两个概念就是`DataSet` 和 `DataStream`,用来表示数据。

## 2. 剖析 Flink程序 ##

每一个Flink程序都包含以下基本的部分

1.  Obtain an `execution environment`, 获取一个执行环境
2.  Load/create the initial data, 加载/创建初始化的数据
3.  Specify transformations on this data,指定data的`transformations`
4.  Specify where to put the results of your computations,执行计算后的结果数据存放位置
5.  Trigger the program execution 触发程序执行

## 3. 延迟计算 ##

All Flink programs are executed lazily: When the program’s main method is executed, the data loading and transformations do not happen directly. Rather, each operation is created and added to the program’s plan. The operations are actually executed when the execution is explicitly triggered by an `execute()` call on the execution environment. Whether the program is executed locally or on a cluster depends on the type of execution environment

所有的Flink程序都是懒执行的,当程序的main method 执行的时候，`加载数据`和`转换`不是立即执行的，而是创建每个操作并将其添加到程序的计划中，当执行环境上的execute（）调用显式触发执行时，才实际执行操作。程序是在本地执行还是在集群上执行取决于执行环境的类型，即在初始化程序创建的environment。

The lazy evaluation lets you construct sophisticated programs that Flink executes as one holistically planned unit.

`懒执行`可以构建Flink作为一个整体计划单元执行的复杂程序。相当于数据库的执行计划。

## 4. key 的 定义方式 ##

Some transformations (join, coGroup, keyBy, groupBy) require that a key be defined on a collection of elements. Other transformations (Reduce, GroupReduce, Aggregate, Windows) allow data being grouped on a key before they are applied.

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

A DataSet is grouped as

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

while a key can be specified on a DataStream using

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

> **NOTE:** In the following discussion we will use the `DataStream` API and `keyBy`. For the DataSet API you just have to replace by `DataSet` and `groupBy`.
> 
> 在`DataStream` 使用`keyBy`;`DataSet` 使用 `groupBy`

### 定义元组的key ###

The simplest case is grouping Tuples on one or more fields of the Tuple:

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

The tuples are grouped on \*\*the first field \*\*(the one of Integer type).

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

Here, we group the tuples on a composite key consisting of **the first and the second field**.

A note on nested Tuples: If you have a DataStream with a nested tuple, such as:

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

Specifying `keyBy(0)` will cause the system to use the full `Tuple2` as a key (with the Integer and Float being the key). If you want to “navigate” into the nested `Tuple2`, you have to use field expression keys which are explained below.

如果想定义嵌套元组里面的值作为key，需要使用 `字段表达式`

### 使用字段表达式定义key ###

**Field Expression Example**:

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

These are valid field expressions for the example code above:

*  `"count"`: The count field in the `WC` class.
 *  `"complex"`: Recursively selects all fields of the field complex of POJO type `ComplexNestedClass`.
 *  `"complex.word.f2"`: Selects the last field of the nested `Tuple3`.
 *  `"complex.hadoopCitizen"`: Selects the Hadoop `IntWritable` type.

### 使用KeySelector 定义Key ###

An additional way to define keys are “key selector” functions. A key selector function takes a single element as input and returns the key for the element. The key can be of any type and be derived from deterministic computations.

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

The following example shows a key selector function that simply returns the field of an object:

这个例子是一个简单的`KeySelector`

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

## 5. 转换函数 ##

Most transformations require user-defined functions. This section lists different ways of how they can be specified

#### Implementing an interface ####

The most basic way is to implement one of the provided interfaces:

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

#### Anonymous classes ####

You can pass a function as an anonymous class:

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

#### Java 8 Lambdas ####

Flink also supports Java 8 Lambdas in the Java API.

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

#### Rich functions ####

All transformations that require a user-defined function can instead take as argument a *rich* function. For example, instead of

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

you can write

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

and pass the function as usual to a `map` transformation:

data.map(new MyMapFunction());

Rich functions can also be defined as an anonymous class:

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

Rich functions provide, in addition to the user-defined function (map, reduce, etc), four methods: `open`, `close`, `getRuntimeContext`, and `setRuntimeContext`.

These are useful for parameterizing the function (see [Passing Parameters to Functions][]), creating and finalizing local state, accessing broadcast variables (see [Broadcast Variables][]), and for accessing runtime information such as accumulators and counters (see [Accumulators and Counters][]), and information on iterations (see [Iterations][]).

## 6. 支持的数据类型 ##

There are six different categories of data types:

1.  **Java Tuples** and **Scala Case Classes**
2.  **Java POJOs**
3.  **Primitive Types**
4.  **Regular Classes**
5.  **Values**
6.  **Hadoop Writables**
7.  **Special Types**

## 7 累加器&计数器 ##

Accumulators are simple constructs with an **add operation** and a **final accumulated result**, which is available after the job ended.

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

The most straightforward accumulator is a **counter**: You can increment it using the `Accumulator.add(V value)` method. At the end of the job Flink will sum up (merge) all partial results and send the result to the client. Accumulators are useful during debugging or if you quickly want to find out more about your data.

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

Flink currently has the following **built-in accumulators**. Each of them implements the [Accumulator][] interface.

*  [**IntCounter**][IntCounter], [**LongCounter**][LongCounter] and [**DoubleCounter**][DoubleCounter]: See below for an example using a counter.
 *  [**Histogram**][Histogram]: A histogram implementation for a discrete number of bins. Internally it is just a map from Integer to Integer. You can use this to compute distributions of values, e.g. the distribution of words-per-line for a word count program.
    
    离散数量的箱的直方图实现。在内部，它只是一个从Integer到Integer的映射。您可以使用它来计算值的分布，例如字数统计程序的每行字数分布。

**How to use accumulators:**

First you have to create an accumulator object (here a counter) in the user-defined transformation function where you want to use it.

private IntCounter numLines = new IntCounter();

Second you have to register the accumulator object, typically in the `open()` method of the *rich* function. Here you also define the name.

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

You can now use the accumulator anywhere in the operator function, including in the `open()` and `close()` methods.

this.numLines.add(1);

The overall result will be stored in the `JobExecutionResult` object which is returned from the `execute()` method of the execution environment (currently this only works if the execution waits for the completion of the job).

myJobExecutionResult.getAccumulatorResult("num-lines")

All accumulators share a single namespace per job. Thus you can use the same accumulator in different operator functions of your job. Flink will internally merge all accumulators with the same name.

**Custom accumulators:**

To implement your own accumulator you simply have to write your implementation of the Accumulator interface. Feel free to create a pull request if you think your custom accumulator should be shipped with Flink.

You have the choice to implement either [Accumulator][] or [SimpleAccumulator][].

`可以理解为一个上下文`

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