Flink基本API及核心概念详解

冷不防 2023-02-11 04:41 136阅读 0赞

## 一.概述 ##

Flink程序是常规程序，可对分布式集合进行转换（例如，过滤，映射，更新状态，联接，分组，定义窗口，聚合）。集合最初是从源创建的（例如，通过读取文件，kafka主题或本地内存中的集合）。结果通过接收器返回，接收器可以将数据写入（分布式）文件或标准输出（例如，命令行终端）。Flink程序可以在各种上下文中运行，独立运行或嵌入其他程序中。执行可以在本地JVM或许多计算机的群集中进行。

根据数据源的类型（即有界或无界源），您将编写批处理程序或流程序，其中DataSet API用于批处理，而DataStream API用于流。本指南将介绍两个API共有的基本概念，但是请参考*流式传输指南* 和 *批处理指南*，以获取有关使用每个API编写程序的具体信息。

## 二.数据集和数据流 ##

Flink具有特殊的类DataSet，DataStream用于表示程序中的数据。可以将它们视为包含重复项的不可变数据集合。DataSet数据是有限的，对于DataStream许多元素，它们可以是无限的。

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

集合最初通过Flink程序添加源创建或从别的集合转换而来，通过使用API方法转化它们衍生的map，filter等等。

## 三.Flink程序剖析 ##

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

1.  获得execution environment
2.  加载/创建初始数据
3.  指定对此数据的转换
4.  指定将计算结果放在何处
5.  触发程序执行

请注意，在包org.apache.flink.api.java中可以找到Java DataSet API的所有核心类， 而在org.apache.flink.streaming.api中可以找到Java DataStream API的类 。

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

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

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

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

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

这将提供一个DataStream，然后可以在其上应用转换以创建新的派生DataStream。  
可以通过使用转换函数在DataStream上调用方法来应用转换。例如，转换如下所示：

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

通过将原始集合中的每个String转换为Integer，将创建一个新的DataStream。  
一旦有了包含最终结果的DataStream，就可以通过创建接收器将其写入外部系统。这些只是创建接收器的一些示例方法：

writeAsText(String path)
    print()

对于指定的完整程序，需要执行程序调用 execute()触发执行StreamExecutionEnvironment。根据ExecutionEnvironment执行类型的不同，执行将在本地计算机上触发或程序提交的集群上执行。

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

## 四.懒加载 ##

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

懒加载使您可以构建复杂的程序，Flink将其作为一个整体计划的单元执行。

## 五.键值对 ##

某些转换【join，coGroup，keyBy，groupBy】要求在dataSet上定义键。其他转换【Reduce，GroupReduce，Aggregate，Windows】允许在应用键之前将数据分组。

dataSet分组操作：

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

dataStream分组操作：

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

Flink的数据模型不是基于键值对。因此，无需将数据集类型实际打包到键和值中。键是“虚拟的”：将它们定义为对实际数据的功能，以指导分组操作。  
1.元组键值对操作  
最简单的情况是在元组的一个或多个字段上对元组进行分组。  
1.1根据元组的第一个字段（整数类型之一）进行分组。

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

1.2将元组分组在由第一字段和第二字段组成的复合键上。

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

1.3嵌套元组

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

指定keyBy(0)将导致系统将完整字符Tuple2用作键（以Integer和Float作为键）。如果要指向到嵌套中Tuple2，则必须使用字段表达式匹配键。

1.4字段表达式定义键  
可以使用基于字符串的字段表达式来引用嵌套字段，并定义用于分组，排序，联接或联合分组的键。

字段表达式使选择（嵌套）复合类型（例如Tuple和POJO类型）中的字段变得非常容易。

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

case class WC(word : String, num : Int) // 样例类
    val counts = text.flatMap(_.split(" ").filter(_.nonEmpty))
        .map(WC(_, 1))
        .groupBy("word")//根据第word进行分组
        .sum("num") // 分组求和
        .setParallelism(1)  // 设置并行度
        .sortPartition("num", Order.DESCENDING) // 降序排序

执行结果：  
![在这里插入图片描述][20200522143015978.png]  
字段表达式语法：

通过字段名称选择POJO字段。例如，"user"引用POJO类型的“用户”字段。

通过其字段名称或字段索引选择元组字段。

可以在POJO和元组中选择嵌套字段。使用"\_“或”."。例如，“user.zip"引用存储在POJO类型的“用户”字段中的POJO的“ zip”字段。支持POJO和元组的任意嵌套和混合，例如”\_2.user.zip"或"user.\_4.1.zip"。

字段表达式示例：

class WC(var complex: ComplexNestedClass, var count: Int) { 
      def this() {  this(null, 0) }
    }
    
    class ComplexNestedClass(
        var someNumber: Int,
        someFloat: Float,
        word: (Long, Long, String),
        hadoopCitizen: IntWritable) { 
      def this() {  this(0, 0, (0, 0, ""), new IntWritable(0)) }
    }

## 六.转换算子 ##

大多数转换算子需要用户定义具体的功能。本节列出了如何指定它们的不同方法：  
1.filter

.filter(_.endsWith("k")) // 过滤

执行结果：  
![在这里插入图片描述][2020052216014025.png]  
2.reduce

val counts = text.flatMap(_.split(" ").filter(_.nonEmpty))
            .map(_ => 1)
          .reduce{ (x, y) => x + y}

执行结果：  
![在这里插入图片描述][20200522170604375.png]  
3.map

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

应用：

val counts = text.flatMap(_.split(" ").filter(_.nonEmpty))
          .map(new MyMapFunction())
          .reduce{ (x, y) => x + y}

执行结果：  
![在这里插入图片描述][20200522183411574.png]  
也可以定义为匿名类：

val counts = text.flatMap(_.split(" ").filter(_.nonEmpty))
          .map(new MyMapFunction())
          .map (new RichMapFunction[Int, Int] {  // 匿名类
            def map(in: Int):Int = {  in + 1 }
          })
          .reduce{ (x, y) => x + y}

执行结果：  
![在这里插入图片描述][20200522191208867.png]

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

Flink对可以在DataSet或DataStream中的元素类型设置了一些限制。原因是系统分析类型以确定有效的执行策略。数据类型有七种不同的类别：

*  Java Tuples and Scala Case Classes  
    元组是复合类型，包含固定数量的各种类型的字段。Java API提供了从Tuple1到的类Tuple25。元组的每个字段可以是任意Flink类型，包括其他元组，从而导致嵌套元组。可以使用字段名称as 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")

*  Java POJOs  
    略，下面会详细讲解
 *  Primitive Types基本类型  
    Flink支持所有Java和Scala的原始类型，如Integer，String和Double。
 *  Regular Classes常规类类型  
    Flink支持大多数Java和Scala类（API和自定义）。除包含无法序列化的字段的类外，例如文件指针，I / O流或其他本机资源。遵循Java Bean约定的类通常可以很好地工作。  
    所有未标识为POJO类型的类（请参见下面的POJO要求）都由Flink处理为常规类类型。Flink将这些数据类型视为黑盒，并且无法访问它们的内容（即，进行有效排序）。通用类型使用序列化框架Kryo进行反序列化。
 *  Values值类型  
    值类型需要手动实现其序列化和反序列化。它们没有通用的序列化框架，而是通过org.apache.flinktypes.Value使用方法read和实现接口为这些操作提供了自定义代码write。当通用序列化效率非常低时，使用Value类型是合理的。一个示例是将元素的稀疏向量实现为数组的数据类型。知道数组大部分为零，就可以对非零元素使用一种特殊的编码，而通用序列化将只写所有数组元素。  
    该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和Try。Java API具有自己的自定义实现Either。与Scala的类似Either，它表示两种可能的值Left或Right。 Either对于错误处理或需要输出两种不同类型的记录的运算符可能很有用。

## 八.POJO ##

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

*  该类必须是公开的。
 *  它必须具有不带参数的公共构造函数（默认构造函数）。
 *  所有字段都是公共的，或者必须可以通过getter和setter函数访问。对于称为foo getter和setter方法的字段，必须命名为getFoo()和setFoo()。
 *  注册的序列化程序必须支持字段的类型。

POJO样例类：

case class WordWithCount(word : String, count : Int){ 
      def this(){ 
        this(null, 0)
      }
    }

执行代码：

val input = execution.fromElements(
          new WordWithCount("hello", 1),
          new WordWithCount("word", 2))
    
        val result = input
          .groupBy("word")
          .sum("count") // 分组求和
          .setParallelism(1)  // 设置并行度
          .sortPartition("count", Order.DESCENDING) // 降序排序
    
        result.print()

执行结果：  
![在这里插入图片描述][20200523100503630.png]

[20200522143015978.png]: https://img-blog.csdnimg.cn/20200522143015978.png
[2020052216014025.png]: https://img-blog.csdnimg.cn/2020052216014025.png
[20200522170604375.png]: https://img-blog.csdnimg.cn/20200522170604375.png
[20200522183411574.png]: https://img-blog.csdnimg.cn/20200522183411574.png
[20200522191208867.png]: https://img-blog.csdnimg.cn/20200522191208867.png
[20200523100503630.png]: https://img-blog.csdnimg.cn/20200523100503630.png