数据流—DataStreamAPI

以你之姓@ 2023-10-04 16:21 16阅读 0赞

## Hello Flink ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70]

## 1：构建一个典型的Flink流式应用需要一下几步： ##

1：设置执行环境。  
2：从数据源中读取一条或多条流  
3：通过一系列流式转换来实现应用逻辑。  
4：选择性的将结果输出到一个或多个数据汇（用来接收处理结果并发送）中  
5:执行程序

### 1：设置执行环境。 ###

执行环境决定了应用是在本地机器上还是集群上运行。  
DataStream API的**执行环境**由：StreamExecutionEnviroment来表示  
调用静态方的getExecutionEnvironment()方法来**执行环境**

根据调用上下文的不同，该方法可能返回一个本地或远程环境。如果一个**连接远程集群的提交客户端**调用了该方法，则会返回一个**远程执行环境**；否则返回一个**本地环境**。

//创建一个本地的流式执行环境
    val localEnv:StreamExecutionEnvironment.createLocalEnvironment()
    //创建一个远程的流式执行环境
    val remoteEnv=StreamExecutionEnvironment.ctreatRemoteEnvirnment(
    "host",
    1234,
    "path/to/jarFile.jar")

接下来使用env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)指定程序采用事件时间

### 2：从数据源中读取一条或多条 ###

### 读取输入流 ###

SteamExecutionEnvironment：提供一系列**创建时数据源**的**方法**，用以将**数据流读取**到引用中。

这些数据流的来源可以是消息队列或文件，也可以是实时生成的。

val sensorData:DataStream[SensorReading] =env.addSource(new SensorSource)

连接数据源，并创建出SensorReading类型的初始DataStream（数据源）  
每一条SensorReading数据都包含了传感器ID，测量时间以及温度

assignTimestampsAndWatermarks(new SensorTimeAssigner):负责分配时间时间所需要的时间戳和水位线

### 3：通过一系列流式转换来实现应用逻辑。 ###

### 应用转换 ###

1：转换的类型多种多样：有些会@1生成一个新的DataStream（类型可能会发生变化）；而@2另外的一些不会修改 DataStream 中的记录，仅会通过分区或分组的方式将其重新组织。

利用**map（）转换**将每个传感器的温度都转换为摄氏度，然后我们使用keyBy()转换，将**传感器读数按照传感器ID**进行**分区**，接下来timeWindow（）转换，针对每个传感器ID分区都将读数划分为5秒一次的滚动窗口：

val avgTemp:DataStream[SensorReading]=sensorData
    	.map(r =>{
        
    			val celsius = (r.temperaure - 32)*(5.0/9.0)
    			SensorReading(r.id,r.timestamp,celsius)
    	})
    	.keyBy(_.id)
    	.timeWindow(Time.seconds(5))
    	.apply(new TemperatureAverager)

map():传感器温度–>摄氏度  
keyBy():将传感器读书按照ID进行分区。  
timeWindow()：针对每个传感器ID分区都划分为5s一次的滚动窗口  
最后使用一个用户定义函数（user-defined function）来计算每个窗口平均温度。

### 4：选择性的将结果输出到一个或多个数据汇（用来接收处理结果并发送）中 ###

流式应用通常会把结果发送到某些外部系统

Flink提供了一组维护状态良好的流式数据汇可以完成上述工作，也可以选择自己实现流式数据汇。还有一些应用不会发出结果，而是将他们保存在**内部**。利用FLink的可查询式状态功能提供服务。

会将DataFrame\[SensorReading\]中的记录作为结果输出。  
每条记录都会包对应传感器在5秒内的平均温度

avgTemp.print()

注意：无论应用结果是**至少一次语义**还是**精确一次语义**，流式数据汇的选择**都将影响应用端到端的一致性**，改一致性取决于所选数据汇和Flink检查点算法的组合情况

### 5:执行程序 ###

应用完成调\#\# 标题用StreamExecutionEnvironment.execute()来执行它。

env.execute("Compute average sensor temperature")

Flink程序都是通过延迟计算（lazily execute）的方式执行。  
哪些创建数据源和转换操作的API调用不会立即触发数据处理，而只会执行环境中构建的一个执行计划。计划中包含了从环境的创建的流式数据源以及应用于这些数据源的转换。只有调用execute()方法是，系统才会触发程序执行。

构建完的计划会被转成 **JobGraph并提交至 JobManager执行**。

根据执行环境类型的不同，系统可能需要将JobGraph发送到作为本地线程启动的JM个（本地执行环境），也可能会将其发送到远程JM上。如果是后者，除了JG外，还需要同时提供包含应用所需全部类和依赖的JAR包。

## 转换操作 ##

流式转换：以一个或多个数据流为输入，并将他们转换为一个或多个的输出流。  
完成一个DS API程序，本质上：通过组合不同的转换来创建一个满足引用逻辑的的DataFlow图

大多数流式转换基于用户自定义函数完成

大多数函数接口设计为\*\*SAM（single abstract method 单一抽象方法）\*\*形式的

为了简介我们会使用lambda函数而不是函数类

### **四类用于数据流转换的API：** ###

### 1. 作用于单个事件的基本转换。 ###

**意味着：每条输出记录都由单个输入记录所生成**

### 基本转换： ###

\*\*1：Map（）：\*\*通过调用 DataStream.map （）方法可以指定map转换产生一个新的DataStream 。

该转换将每个到来的事件传给一个用户自定义的映射器（user defined mapper），后者针对每一个输入只会返回一个输出事件（可能类型发生改变的输出事件）。

MapFunction的两个类型的参数分别是输出事件与输入事件的类型，他们可以通过MapFunction接口来指定。该接口的Map()方法将每一个输入事件转换为输出事件  
![在这里插入图片描述][20210614135230798.png]

val reading: DataStream[SensorReading]=...
    val sensorIds: DS[String] =reading.map(r=>r.id)

**2：Filter（）：** filter转换利用一个作用在流中每条输入事件上的 **布尔条件来决定事件的去留**：  
如果返回值为 true ，那么它会保留输入事件井将其转发到输出，否则它会把事件丢弃。

通过调用DataStream.filter()方法可以指定filter转换**产生一个数据类型不变的DS**  
![在这里插入图片描述][20210614141609518.png]  
![在这里插入图片描述][20210614141640274.png]  
**3:FlatMap（）：** flatMap转换类似于map，但它可以对每个输入事件产生零个、一个或多个输出事件。事实上，**flatMap转换可以看做是filter和map的泛化，它能够实现后两者的操作**。

![在这里插入图片描述][20210614142821195.png]  
flatMap函数会作用于一个语句流上，将**每个语句按照空格字符分割**，然后**把分割得到的每个单词作为一条独立的记录发送出去**。

val sentences : DataStream[String]=...
    val words: DataStream[String]=sentences
    .flatmap(id=>id.split(""))

### 2. 针对相同键值事件的KeyedStream转换。 ###

很多应用需要**将事件按照某个属性分组后再进行处理**。  
作为DataStream API中一类特殊的DataStream,KeyedStream抽象可以**从逻辑上**将**事件按照键值** **分配**到**多条独立的子流**中。

作用于KeyedStream的状态化转换可以**对当前处理事件的键值所对应的上下文中的状态进行读写**。这意味着**所有键值相同的时间可以访问相同的状态**，因此**可以一并处**理

注意：在**使用状态化转**换和**基于键值的聚合**时**要小心**。如果**键值域（key domain）会持续增长**（例如：将唯一的事物ID作为键值），则必须对那些**不在活跃的键值进行清理**，**以避免出现内存问题。**

keyedStream也可以使用map，filter，flatmap等转换进行处理

**1：keyBy：转换DS为KeyedStream,然后对它进行滚动聚合以及reduce操作**  
keyBy转换通过指定键值的方式将**一个DataStream转化为KeyedStream**。流中的事件会根据各自键值被分到不同的分区，这样一来，**有着相同键值的事件一定会在后续算子的同一个任务上处理。**，（主要实现功能的函数：keyBy）

经过keyBy处理后，一条数据流分两类，以颜色作为键值，下图中将黑色事件分到一个任务上，而将其他时间分到另一个任务上  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 1]  
keyBy（）方法接收一个用来指定分区键值（可以是多个）的参数，返回一个KeyedStream。  
lambda函数r =>r.id表示从传感器读数r中提取id字段  
![在这里插入图片描述][20210614150105814.png]  
**2：滚动聚合**  
转换作用于KeyedStream上，它将生成一个包含聚合结果（例如求和、最小值、最大值等）的DataStream。该动聚合算子会对每一个遇到过的键值保存一个聚合结果。

滚动聚合虽然不需要用户自定义函数，但需要接收一个用于指定聚合目标字段的参数。

DataStreamAPI中提供了以下滚动聚合方法：  
sum() 滚动计算输入流中指定字段的和。  
min() 滚动计算输入流中指定字段的最小值。  
max() 该动计算输入流中指定字段的最大值。  
minBy()滚动计算输入流中迄今为止最小值，**返回该值所在事件**。  
maxBy() 滚动计算输入流中迄今为止最大值，**返回该值所在事件**。

请注意，你无法将多个滚动聚合方法组合使用，**每次只能计算一个。**  
以下例子中，我们对一个Tuple3\[Int,Int, Int］类型的数据流按照第一个字段进行键值分区，然后滚动计算第二个字段的总和：

示例对于：键值为1的首先输出（1,2，2）然后（1,7，2）：找到键值第一个值为1的值，然后进行下标为1 的求和，替代第一次输出的值，第三个数不变  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 2]  
**滚动聚合值对有限键值使用：**  
由于滚动聚合算子会为每个处理过的键值维持一个状态。由于这些状态不会被自动清理，所以该算子只能用于键值域有限的流。

**3：reduce转换是滚动聚合转换的泛化。**  
它将一个ReduceFunction应用在一个KeyedStream上，**每个到来事件都会和reduce结果进行一次组合**，从而产生一个新的DataStream。reduce转换不会改变数据类型，因此输出流的类型会永远和输入流保持一致。

下面示例中，数据流会以语言字段（即第一个字段）为键值进行分区，最终结果是针对每种语言产生一个不断更新的单词列表：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 3]

用于reduce的lambda函数会直接转发第一个字段（键值字段），并将所有第二个字段中List【String】值连接合起来，下一次只传一次，传的是上一次的计算结果  
**滚动reduce操作只对有限键值使用：**  
由于滚动reduce算子会为每个处理过的键值维持一个状态。由于这些状态不会被自动清理，所以该算子只能用于键值域有限的流。

### 3. 将多条数据流合并为一条或将一条数据流拆分成多条流的转换。、 ###

### 多流转换 ###

很多应用需要将多条输入流联合起来处理，或将一条流分割成多条子流以应用不同逻辑。  
哪些能同时处理多条输入流或产生多条结果流的DataStream API转换？

**1:Union**  
DataStream.union（）方法可以合并两条或多条类型相同的DataStream，生成一个新的类型相同的DataStream。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 4]

Union  
union执行过程中，来自两条流的事件会以FIFO（先进先出）的方式合井，其顺序无住得到任何保证。此外，union算子不会对数据进行去重，每个输入消息都会被发往下游算子。

以下示例展示了如何将三条类型为SensorReading的数据流合并为一条：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 5]  
**2:Connect（只能连接两个）**  
有一个森林区域监控应用会在火灾发生风险很高时报警。该应用会接收一条包含之前所见到的全部温度传感器读数的数据流，以及另外一条烟雾指数测量值数据流。当温度超过给定阈值且烟雾指数很高时，应用就会发出火灾警报。  
DataStream API提供的connect转换可以用来实现该用例。DataStream.connect（）方法接收一个DataStream并返回一个ConnectedStream对象，该对象表示两个联结起来（connected）的流：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 6]  
**3:Split和Select**  
split转换是union转换的逆操作。它将输入流分割成两条或多条类型和输入流相同的输出流。每一个到来的事件都可以被发往零个、一个或多个输出流。因此，split也可以用来过滤或复制事件。

DataStream.split（）方法会返回一个SplitStream对象，它提供的select()方法可以让我们通过指定输出名称的方式从SplitStream中选择一条或多条流。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 7]  
Split和Select  
注：在Flink DataStream api中有一个split()算子，它的功能是将一个DataStream，通过split()设置多个标记，划分成多个流。再通过select()获取对应标记的流。  
将一条数字流分成一条大数字流和一条小数字流。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 8]

### 4. 对流中的事件进行重新组织的分发转换 ###

Flink中的各类分区转换对应我们在第2章的“数据交换策略”中介绍的多种数据交换策略。这些操作定义了如何将事件分配给不同任务。在使用DataStream API构建程序时，系统会根据操作语义和配置的并行度自动选择数据分区策略并将数据转发到正确的目标。有时也需要自定义分区策略或自定义分区器。

### DataStream中用于**控制分区策略**或**自定义分区策略**的方法如下。 ###

\*\*随机：\*\*我们可以利用DataStream.shuffle（）方法实现随机数据交换策略。  
**轮流**：rebalance（）方法会将输入流中的事件以轮流方式均匀分配给后继任务  
**重调：** rescale（）也会以轮流方式对事件进行分发，但分发目标仅限于部分后继任务。  
rebalance()和rescale区别：  
rebalance（）和rescale（）的本质不同体现在生成任务连接的方式。rebalance（）会在所有发送任务和接收任务之间建立通信通道；而rescale（）中每个发送任务只会和下游算子的部分任务建立通道

**广播**：broadcast （）方法会将输入流中的事件复制并发往所有下游算子的并行任务。

\*\*全局：\*\*global（）方法会将输入流中的所有事件发往下游算子的第一个并行任务。使用此分区策略时务必小心，因为将所有事件发往同一任务可能会影响程序性能。

\*\*自定义：\*\*如果所有预定义的分区策略都不合适，你可以利用partitionCustom（）方法自己定义分区策略。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 9]

### 设置并行度 ###

**类型**  
**1:原始类型:Int(java中Integer),String,Double  
2:java和scala元祖  
3：scala样例类  
4：POJO（包括Apache Avro生成的类）：对象存储，序列化、是将对象的状态信息转换为可以存储或传输的形式的过程。**  
Flink会分析那些不属于任何一类的数据类型，并尝试将它们作为POJO类型进行处理。如果一个类满足如下条件，Flink就会将它看做POJO(序列化方式不同）

是一个公有类。  
有一个公有的无参默认构造函数。  
所有字段都是公有的或提供了相应的getter及setter方住。这些方法需要遵循默认的命名规范，即对于Y类型的x字段方住头分别为Y getX（）和setX(Yx）。  
所有宇段类型都必须是Flink所支持的。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210613094825501.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3,size_16,color_FFFFFF,t_70
[20210614135230798.png]: https://img-blog.csdnimg.cn/20210614135230798.png
[20210614141609518.png]: https://img-blog.csdnimg.cn/20210614141609518.png
[20210614141640274.png]: https://img-blog.csdnimg.cn/20210614141640274.png
[20210614142821195.png]: https://img-blog.csdnimg.cn/20210614142821195.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210614142921837.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3,size_16,color_FFFFFF,t_70
[20210614150105814.png]: https://img-blog.csdnimg.cn/20210614150105814.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210614150406485.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20210614151648446.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20210614152432266.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20210614152446929.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/2021061415252071.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20210614152542871.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20210614152559182.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20210614152659334.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NjU2MDc3,size_16,color_FFFFFF,t_70