# Flink1.17_source_notes
**Repository Path**: jiongmefeishi/flink1.17_source_notes
## Basic Information
- **Project Name**: Flink1.17_source_notes
- **Description**: 尚硅谷大数据Flink1.17实战教程从入门到精通（基础篇、核心篇、高阶篇、SQL篇）
笔记 & 源码
b站视频地址：https://www.bilibili.com/video/BV1eg4y1V7AN/?vd_source=580fae2c111643f55c1dec0f5d0837f6
- **Primary Language**: Java
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No
## Statistics
- **Stars**: 0
- **Forks**: 2
- **Created**: 2024-08-29
- **Last Updated**: 2024-08-29
## Categories & Tags
**Categories**: Uncategorized
**Tags**: None
## README
# 大数据技术之Flink
## **第1章 Flink概述**
### **1.1 Flink是什么**
### **1.2 Flink特点**
### **1.3 Flink vs SparkStreaming**
表 Flink 和 Streaming对比
| | **Flink** | **Streaming** |
| ------------ | ------------------ | ------------------------------------ |
| **计算模型** | 流计算 | 微批处理 |
| **时间语义** | 事件时间、处理时间 | 处理时间 |
| **窗口** | 多、灵活 | 少、不灵活（窗口必须是批次的整数倍） |
| **状态** | 有 | 没有 |
| **流式SQL** | 有 | 没有 |
### **1.4 Flink的应用场景**
### **1.5 Flink分层API**
## 第2章 Flink快速上手
### **2.1 创建项目**
在准备好所有的开发环境之后，我们就可以开始开发自己的第一个Flink程序了。首先我们要做的，就是在IDEA中搭建一个Flink项目的骨架。我们会使用Java项目中常见的Maven来进行依赖管理。
**1）创建工程**
（1）打开IntelliJ IDEA，创建一个Maven工程。
（2）将这个Maven工程命名为FlinkTutorial。
（3）选定这个Maven工程所在存储路径，并点击Finish，Maven工程即创建成功。
**2）添加项目依赖**
在项目的pom文件中，添加Flink的依赖，包括flink-java、flink-streaming-java，以及flink-clients（客户端，也可以省略）。
```xml
1.17.0
org.apache.flink
flink-streaming-java
${flink.version}
org.apache.flink
flink-clients
${flink.version}
```
### **2.2 WordCount代码编写**
需求：统计一段文字中，每个单词出现的频次。
环境准备：在src/main/java目录下，新建一个包，命名为com.atguigu.wc。
#### **2.2.1 批处理**
批处理基本思路：先逐行读入文件数据，然后将每一行文字拆分成单词；接着按照单词分组，统计每组数据的个数，就是对应单词的频次。
**1）数据准备**
（1）在工程根目录下新建一个input文件夹，并在下面创建文本文件words.txt
（2）在words.txt中输入一些文字，例如：
```shell
hello flink
hello world
hello java
```
**2）代码编写**
（1）在com.atguigu.wc包下新建Java类BatchWordCount，在静态main方法中编写代码。具体代码实现如下：
```java
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
public class BatchWordCount {
public static void main(String[] args) throws Exception {
// 1. 创建执行环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
// 2. 从文件读取数据 按行读取(存储的元素就是每行的文本)
DataSource lineDS = env.readTextFile("input/words.txt");
// 3. 转换数据格式
FlatMapOperator> wordAndOne = lineDS.flatMap(new FlatMapFunction>() {
@Override
public void flatMap(String line, Collector> out) throws Exception {
String[] words = line.split(" ");
for (String word : words) {
out.collect(Tuple2.of(word,1L));
}
}
});
// 4. 按照 word 进行分组
UnsortedGrouping> wordAndOneUG = wordAndOne.groupBy(0);
// 5. 分组内聚合统计
AggregateOperator> sum = wordAndOneUG.sum(1);
// 6. 打印结果
sum.print();
}
}
```
（2）输出
```shell
(flink,1)
(world,1)
(hello,3)
(java,1)
```
需要注意的是，这种代码的实现方式，是基于DataSet API的，也就是我们对数据的处理转换，是看作数据集来进行操作的。事实上Flink本身是流批统一的处理架构，批量的数据集本质上也是流，没有必要用两套不同的API来实现。所以从Flink 1.12开始，官方推荐的做法是直接使用DataStream API，在提交任务时通过将执行模式设为BATCH来进行批处理：
$ bin/flink run -Dexecution.runtime-mode=BATCH BatchWordCount.jar
这样，DataSet API就没什么用了，在实际应用中我们只要维护一套DataStream API就可以。这里只是为了方便大家理解，我们依然用DataSet API做了批处理的实现。
#### **2.2.2 流处理**
对于Flink而言，流才是整个处理逻辑的底层核心，所以流批统一之后的DataStream API更加强大，可以直接处理批处理和流处理的所有场景。
下面我们就针对不同类型的输入数据源，用具体的代码来实现流处理。
**1）读取文件**
我们同样试图读取文档words.txt中的数据，并统计每个单词出现的频次。整体思路与之前的批处理非常类似，代码模式也基本一致。
在com.atguigu.wc包下新建Java类StreamWordCount，在静态main方法中编写代码。具体代码实现如下：
```java
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import java.util.Arrays;
public class StreamWordCount {
public static void main(String[] args) throws Exception {
// 1. 创建流式执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2. 读取文件
DataStreamSource lineStream = env.readTextFile("input/words.txt");
// 3. 转换、分组、求和，得到统计结果
SingleOutputStreamOperator> sum = lineStream.flatMap(new FlatMapFunction>() {
@Override
public void flatMap(String line, Collector> out) throws Exception {
String[] words = line.split(" ");
for (String word : words) {
out.collect(Tuple2.of(word, 1L));
}
}
}).keyBy(data -> data.f0)
.sum(1);
// 4. 打印
sum.print();
// 5. 执行
env.execute();
}
}
```
输出：
```shell
3> (java,1)
5> (hello,1)
5> (hello,2)
5> (hello,3)
13> (flink,1)
9> (world,1)
```
主要观察与批处理程序BatchWordCount的不同：
- 创建执行环境的不同，流处理程序使用的是StreamExecutionEnvironment。
- 转换处理之后，得到的数据对象类型不同。
- 分组操作调用的是keyBy方法，可以传入一个匿名函数作为键选择器（KeySelector），指定当前分组的key是什么。
- 代码末尾需要调用env的execute方法，开始执行任务。
**2）读取socket文本流**
在实际的生产环境中，真正的数据流其实是无界的，有开始却没有结束，这就要求我们需要持续地处理捕获的数据。为了模拟这种场景，可以监听socket端口，然后向该端口不断的发送数据。
（1）将StreamWordCount代码中读取文件数据的readTextFile方法，替换成读取socket文本流的方法socketTextStream。具体代码实现如下：
```java
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import java.util.Arrays;
public class SocketStreamWordCount {
public static void main(String[] args) throws Exception {
// 1. 创建流式执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2. 读取文本流：hadoop102表示发送端主机名、7777表示端口号
DataStreamSource lineStream = env.socketTextStream("hadoop102", 7777);
// 3. 转换、分组、求和，得到统计结果
SingleOutputStreamOperator> sum = lineStream.flatMap((String line, Collector> out) -> {
String[] words = line.split(" ");
for (String word : words) {
out.collect(Tuple2.of(word, 1L));
}
}).returns(Types.TUPLE(Types.STRING, Types.LONG))
.keyBy(data -> data.f0)
.sum(1);
// 4. 打印
sum.print();
// 5. 执行
env.execute();
}
}
```
（2）在Linux环境的主机hadoop102上，执行下列命令，发送数据进行测试
```shell
[atguigu@hadoop102 ~]$ nc -lk 7777
```
注意：要先启动端口，后启动StreamWordCount程序，否则会报超时连接异常。
（3）启动StreamWordCount程序
我们会发现程序启动之后没有任何输出、也不会退出。这是正常的，因为Flink的流处理是事件驱动的，当前程序会一直处于监听状态，只有接收到数据才会执行任务、输出统计结果。
（4）从hadoop102发送数据
①在hadoop102主机中，输入“hello flink”，输出如下内容
```shell
13> (flink,1)
5> (hello,1)
```
②再输入“hello world”，输出如下内容
```shell
2> (world,1)
5> (hello,2)
```
说明：
Flink还具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息，从而获得对应的序列化器和反序列化器。但是，由于Java中泛型擦除的存在，在某些特殊情况下（比如Lambda表达式中），自动提取的信息是不够精细的——只告诉Flink当前的元素由“船头、船身、船尾”构成，根本无法重建出“大船”的模样；这时就需要显式地提供类型信息，才能使应用程序正常工作或提高其性能。
因为对于flatMap里传入的Lambda表达式，系统只能推断出返回的是Tuple2类型，而无法得到Tuple2。只有显式地告诉系统当前的返回类型，才能正确地解析出完整数据。
## **第3章 Flink部署**
### **3.1** **集群角色**
### **3.2 Flink集群搭建**
#### **3.2.1 集群启动**
**0）集群规划**
表3-1 集群角色分配
| **节点服务器** | **hadoop102** | **hadoop103** | **hadoop104** |
| -------------- | --------------------- | ------------- | ------------- |
| **角色** | JobManagerTaskManager | TaskManager | TaskManager |
具体安装部署步骤如下：
**1）下载并解压安装包**
（1）下载安装包flink-1.17.0-bin-scala_2.12.tgz，将该jar包上传到hadoop102节点服务器的/opt/software路径上。
（2）在/opt/software路径上解压flink-1.17.0-bin-scala_2.12.tgz到/opt/module路径上。
```shell
[atguigu@hadoop102 software]$ tar -zxvf flink-1.17.0-bin-scala_2.12.tgz -C /opt/module/
```
**2）修改集群配置**
（1）进入conf路径，修改flink-conf.yaml文件，指定hadoop102节点服务器为JobManager
```shell
[atguigu@hadoop102 conf]$ vim flink-conf.yaml
```
修改如下内容：
```shell
# JobManager节点地址.
jobmanager.rpc.address: hadoop102
jobmanager.bind-host: 0.0.0.0
rest.address: hadoop102
rest.bind-address: 0.0.0.0
# TaskManager节点地址.需要配置为当前机器名
taskmanager.bind-host: 0.0.0.0
taskmanager.host: hadoop102
```
（2）修改workers文件，指定hadoop102、hadoop103和hadoop104为TaskManager
```shel
[atguigu@hadoop102 conf]$ vim workers
```
修改如下内容：
```shell
hadoop102
hadoop103
hadoop104
```
（3）修改masters文件
```shell
[atguigu@hadoop102 conf]$ vim masters
```
修改如下内容：
```shell
hadoop102:8081
```
（4）另外，在flink-conf.yaml文件中还可以对集群中的JobManager和TaskManager组件进行优化配置，主要配置项如下：
- jobmanager.memory.process.size：对JobManager进程可使用到的全部内存进行配置，包括JVM元空间和其他开销，默认为1600M，可以根据集群规模进行适当调整。
- taskmanager.memory.process.size：对TaskManager进程可使用到的全部内存进行配置，包括JVM元空间和其他开销，默认为1728M，可以根据集群规模进行适当调整。
- taskmanager.numberOfTaskSlots：对每个TaskManager能够分配的Slot数量进行配置，默认为1，可根据TaskManager所在的机器能够提供给Flink的CPU数量决定。所谓Slot就是TaskManager中具体运行一个任务所分配的计算资源。
- parallelism.default：Flink任务执行的并行度，默认为1。优先级低于代码中进行的并行度配置和任务提交时使用参数指定的并行度数量。
关于Slot和并行度的概念，我们会在下一章做详细讲解。
**3）分发安装目录**
（1）配置修改完毕后，将Flink安装目录发给另外两个节点服务器。
```shell
[atguigu@hadoop102 module]$ xsync flink-1.17.0/
```
（2）修改hadoop103的 taskmanager.host
```shell
[atguigu@hadoop103 conf]$ vim flink-conf.yaml
```
修改如下内容：
```shell
# TaskManager节点地址.需要配置为当前机器名
taskmanager.host: hadoop103
```
（3）修改hadoop104的 taskmanager.host
```shell
[atguigu@hadoop104 conf]$ vim flink-conf.yaml
```
修改如下内容：
```shell
# TaskManager节点地址.需要配置为当前机器名
taskmanager.host: hadoop104
```
**4）启动集群**
（1）在hadoop102节点服务器上执行start-cluster.sh启动Flink集群：
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/start-cluster.sh
```
（2）查看进程情况：
[atguigu@hadoop102 flink-1.17.0]$ jpsall
```shell
=============== hadoop102 ===============
4453 StandaloneSessionClusterEntrypoint
4458 TaskManagerRunner
4533 Jps
=============== hadoop103 ===============
2872 TaskManagerRunner
2941 Jps
=============== hadoop104 ===============
2948 Jps
2876 TaskManagerRunner
```
**5）访问Web UI**
启动成功后，同样可以访问对flink集群和任务进行监控管理。
这里可以明显看到，当前集群的TaskManager数量为3；由于默认每个TaskManager的Slot数量为1，所以总Slot数和可用Slot数都为3。
#### **3.2.2** **向集群提交作业**
在上一章中，我们已经编写读取socket发送的单词并统计单词的个数程序案例。本节我们将以该程序为例，演示如何将任务提交到集群中进行执行。具体步骤如下。
**1）环境准备**
在hadoop102中执行以下命令启动netcat。
```shell
[atguigu@hadoop102 flink-1.17.0]$ nc -lk 7777
```
**2）程序打包**
（1）在我们编写的Flink入门程序的pom.xml文件中添加打包插件的配置，具体如下：
```xml
org.apache.maven.plugins
maven-shade-plugin
3.2.4
package
shade
com.google.code.findbugs:jsr305
org.slf4j:*
log4j:*
*:*
META-INF/*.SF
META-INF/*.DSA
META-INF/*.RSA
```
（2）插件配置完毕后，可以使用IDEA的Maven工具执行package命令，出现如下提示即表示打包成功。
```shell
-------------------------------------------------------------------
[INFO] BUILD SUCCESS
-------------------------------------------------------------------
```
打包完成后，在target目录下即可找到所需JAR包，JAR包会有两个，FlinkTutorial-1.0-SNAPSHOT.jar和FlinkTutorial-1.0-SNAPSHOT-jar-with-dependencies.jar，因为集群中已经具备任务运行所需的所有依赖，所以建议使用==FlinkTutorial-1.0-SNAPSHOT.jar==。
**3）在WebUI上提交作业**
（1）任务打包完成后，我们打开Flink的WEB UI页面，在右侧导航栏点击“Submit New Job”，然后点击按钮“+ Add New”，选择要上传运行的JAR包，如下图所示。
JAR包上传完成，如下图所示：
（2）点击该JAR包，出现任务配置页面，进行相应配置。
主要配置==程序入口主类的全类名，任务运行的并行度==，任务运行所需的配置参数和保存点路径等，如下图所示，配置完成后，即可点击按钮“Submit”，将任务提交到集群运行。
（3）任务提交成功之后，可点击左侧导航栏的“Running Jobs”查看程序运行列表情况。
（4）测试
①在socket端口中输入hello
```shell
[atguigu@hadoop102 flink-1.17.0]$ nc -lk 7777
hello
```
②先点击Task Manager，然后点击右侧的192.168.10.104服务器节点
③点击Stdout，就可以看到hello单词的统计
==注意：如果hadoop104节点没有统计单词数据，可以去其他TaskManager节点查看。==
（4）点击该任务，可以查看任务运行的具体情况，也可以通过点击“Cancel Job”结束任务运行。
**4）命令行提交作业**
除了通过WEB UI界面提交任务之外，也可以直接通过命令行来提交任务。这里为方便起见，我们可以先把jar包直接上传到目录flink-1.17.0下
（1）首先需要启动集群。
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/start-cluster.sh
```
（2）在hadoop102中执行以下命令启动netcat。
```shell
[atguigu@hadoop102 flink-1.17.0]$ nc -lk 7777
```
（3）将flink程序运行jar包上传到/opt/module/flink-1.17.0路径。
（4）进入到flink的安装路径下，在命令行使用flink run命令提交作业。
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/flink run -m hadoop102:8081 -c com.atguigu.wc.SocketStreamWordCount ./FlinkTutorial-1.0-SNAPSHOT.jar
```
这里的参数 -m指定了提交到的JobManager，-c指定了入口类。
（5）在浏览器中打开Web UI，[http://hadoop102:8081查看应用执行情况](http://hadoop102:8081查看应用执行情况，)。
用netcat输入数据，可以在TaskManager的标准输出（Stdout）看到对应的统计结果。
（6）在/opt/module/flink-1.17.0/log路径中，可以查看TaskManager节点。
```shell
[atguigu@hadoop102 log]$ cat flink-atguigu-standalonesession-0-hadoop102.out
(hello,1)
(hello,2)
(flink,1)
(hello,3)
(scala,1)
```
### **3.3 部署模式**
在一些应用场景中，对于集群资源分配和占用的方式，可能会有特定的需求。Flink为各种场景提供了不同的部署模式，主要有以下三种：==会话模式（Session Mode）、单作业模式（Per-Job Mode）、应用模式（Application Mode）==。
它们的区别主要在于：集群的==生命周期==以及==资源的分配方式==；以及应用的==main方法到底在哪里执行==——客户端（Client）还是JobManager。
#### **3.3.1 会话模式（Session Mode）**
#### **3.3.2 单作业模式（Per-Job Mode）**
#### **3.3.3 应用模式（Application Mode）**
这里我们所讲到的部署模式，相对是比较抽象的概念。实际应用时，一般需要和资源管理平台结合起来，选择特定的模式来分配资源、部署应用。接下来，我们就针对不同的资源提供者的场景，具体介绍Flink的部署方式。
### **3.4 Standalone运行模式（了解）**
独立模式是独立运行的，不依赖任何外部的资源管理平台；当然独立也是有代价的：如果==资源不足，或者出现故障，没有自动扩展或重分配资源的保证，必须手动处理==。所以独立模式一般==只用在开发测试或作业非常少的场景下==。
#### **3.4.1 会话模式部署**
我们在第3.2节用的就是Standalone集群的会话模式部署。
==提前启动集群，并通过Web页面客户端提交任务（可以多个任务，但是集群资源固定）。==
#### **3.4.2 单作业模式部署**
==Flink的Standalone集群并不支持单作业模式部署。==因为单作业模式需要借助一些资源管理平台。
#### **3.4.3 应用模式部署**
应用模式下不会提前创建集群，所以不能调用start-cluster.sh脚本。我们可以使用同样在bin目录下的standalone-job.sh来创建一个JobManager。
具体步骤如下：
（0）环境准备。在hadoop102中执行以下命令启动netcat。
```shell
[atguigu@hadoop102 flink-1.17.0]$ nc -lk 7777
```
（1）进入到Flink的安装路径下，将应用程序的jar包放到lib/目录下。
```shell
[atguigu@hadoop102 flink-1.17.0]$ mv FlinkTutorial-1.0-SNAPSHOT.jar lib/
```
（2）执行以下命令，启动JobManager。
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/standalone-job.sh start --job-classname com.atguigu.wc.SocketStreamWordCount
```
这里我们直接指定作业入口类，脚本会到lib目录扫描所有的jar包。
（3）同样是使用bin目录下的脚本，启动TaskManager。
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/taskmanager.sh start
```
（4）在hadoop102上模拟发送单词数据。
```shell
[atguigu@hadoop102 ~]$ nc -lk 7777
hello
```
（5）在hadoop102:8081地址中观察输出数据
（6）如果希望停掉集群，同样可以使用脚本，命令如下。
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/taskmanager.sh stop
[atguigu@hadoop102 flink-1.17.0]$ bin/standalone-job.sh stop
```
### **3.5 YARN运行模式（重点）**
YARN上部署的过程是：客户端把Flink应用提交给Yarn的ResourceManager，Yarn的ResourceManager会向Yarn的NodeManager申请容器。在这些容器上，Flink会部署JobManager和TaskManager的实例，从而启动集群。Flink会根据运行在JobManger上的作业所需要的Slot数量动态分配TaskManager资源。
#### **3.5.1 相关准备和配置**
在将Flink任务部署至YARN集群之前，需要确认集群是否安装有Hadoop，保证Hadoop版本至少在2.2以上，并且集群中安装有HDFS服务。
具体配置步骤如下：
（1）配置环境变量，增加环境变量配置如下：
```shell
$ sudo vim /etc/profile.d/my_env.sh
HADOOP_HOME=/opt/module/hadoop-3.3.4
export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin
export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop
export HADOOP_CLASSPATH=`hadoop classpath`
```
（2）启动Hadoop集群，包括HDFS和YARN。
```shell
[atguigu@hadoop102 hadoop-3.3.4]$ start-dfs.sh
[atguigu@hadoop103 hadoop-3.3.4]$ start-yarn.sh
```
（3）在hadoop102中执行以下命令启动netcat。
```shell
[atguigu@hadoop102 flink-1.17.0]$ nc -lk 7777
```
#### **3.5.2 会话模式部署**
YARN的会话模式与独立集群略有不同，需要首先申请一个YARN会话（YARN Session）来启动Flink集群。具体步骤如下：
**1）启动集群**
（1）启动Hadoop集群（HDFS、YARN）。
（2）执行脚本命令向YARN集群申请资源，开启一个YARN会话，启动Flink集群。
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/yarn-session.sh -nm test
```
可用参数解读：
- -d：分离模式，如果你不想让Flink YARN客户端一直前台运行，可以使用这个参数，即使关掉当前对话窗口，YARN session也可以后台运行。
- -jm（--jobManagerMemory）：配置JobManager所需内存，默认单位MB。
- -nm（--name）：配置在YARN UI界面上显示的任务名。
- -qu（--queue）：指定YARN队列名。
- -tm（--taskManager）：配置每个TaskManager所使用内存。
注意：Flink1.11.0版本不再使用-n参数和-s参数分别指定TaskManager数量和slot数量，YARN会按照需求动态分配TaskManager和slot。所以从这个意义上讲，YARN的会话模式也不会把集群资源固定，同样是动态分配的。
YARN Session启动之后会给出一个Web UI地址以及一个YARN application ID，如下所示，用户可以通过Web UI或者命令行两种方式提交作业。
```shell
2022-11-17 15:20:52,711 INFO org.apache.flink.yarn.YarnClusterDescriptor [] - Found Web Interface hadoop104:40825 of application 'application_1668668287070_0005'.
JobManager Web Interface: http://hadoop104:40825
```
**2）提交作业**
（1）通过Web UI提交作业
这种方式比较简单，与上文所述Standalone部署模式基本相同。
（2）通过命令行提交作业
① 将**FlinkTutorial-1.0-SNAPSHOT.jar**任务上传至集群。
② 执行以下命令将该任务提交到已经开启的Yarn-Session中运行。
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/flink run -c com.atguigu.wc.SocketStreamWordCount FlinkTutorial-1.0-SNAPSHOT.jar
```
客户端可以自行确定JobManager的地址，也可以通过-m或者-jobmanager参数指定JobManager的地址，JobManager的地址在YARN Session的启动页面中可以找到。
③ 任务提交成功后，可在YARN的Web UI界面查看运行情况。hadoop103:8088。
从上图中可以看到我们创建的Yarn-Session实际上是一个Yarn的Application，并且有唯一的Application ID。
④也可以通过Flink的Web UI页面查看提交任务的运行情况，如下图所示。
#### **3.5.3 单作业模式部署**
在YARN环境中，由于有了外部平台做资源调度，所以我们也可以直接向YARN提交一个单独的作业，从而启动一个Flink集群。
（1）执行命令提交作业。
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/flink run -d -t yarn-per-job -c com.atguigu.wc.SocketStreamWordCount FlinkTutorial-1.0-SNAPSHOT.jar
```
注意：如果启动过程中报如下异常。
```shell
Exception in thread “Thread-5” java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration ‘classloader.check-leaked-classloader’.
at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders
```
解决办法：在flink的/opt/module/flink-1.17.0/conf/flink-conf.yaml配置文件中设置
```shell
[atguigu@hadoop102 conf]$ vim flink-conf.yaml
classloader.check-leaked-classloader: false
```
（2）在YARN的ResourceManager界面查看执行情况。
点击可以打开Flink Web UI页面进行监控，如下图所示：
（3）可以使用命令行查看或取消作业，命令如下。
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/flink list -t yarn-per-job -Dyarn.application.id=application_XXXX_YY
[atguigu@hadoop102 flink-1.17.0]$ bin/flink cancel -t yarn-per-job -Dyarn.application.id=application_XXXX_YY
```
这里的application_XXXX_YY是当前应用的ID，是作业的ID。注意如果取消作业，整个Flink集群也会停掉。
#### **3.5.4 应用模式部署**
应用模式同样非常简单，与单作业模式类似，直接执行flink run-application命令即可。
**1）命令行提交**
（1）执行命令提交作业。
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/flink run-application -t yarn-application -c com.atguigu.wc.SocketStreamWordCount FlinkTutorial-1.0-SNAPSHOT.jar
```
（2）在命令行中查看或取消作业。
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY
[atguigu@hadoop102 flink-1.17.0]$ bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY
```
**2）上传HDFS提交**
可以通过yarn.provided.lib.dirs配置选项指定位置，将flink的依赖上传到远程。
（1）上传flink的lib和plugins到HDFS上
```shell
[atguigu@hadoop102 flink-1.17.0]$ hadoop fs -mkdir /flink-dist
[atguigu@hadoop102 flink-1.17.0]$ hadoop fs -put lib/ /flink-dist
[atguigu@hadoop102 flink-1.17.0]$ hadoop fs -put plugins/ /flink-dist
```
（2）上传自己的jar包到HDFS
```shell
[atguigu@hadoop102 flink-1.17.0]$ hadoop fs -mkdir /flink-jars
[atguigu@hadoop102 flink-1.17.0]$ hadoop fs -put FlinkTutorial-1.0-SNAPSHOT.jar /flink-jars
```
（3）提交作业
```shell
[atguigu@hadoop102 flink-1.17.0]$ bin/flink run-application -t yarn-application -Dyarn.provided.lib.dirs="hdfs://hadoop102:8020/flink-dist" -c com.atguigu.wc.SocketStreamWordCount hdfs://hadoop102:8020/flink-jars/FlinkTutorial-1.0-SNAPSHOT.jar
```
这种方式下，flink本身的依赖和用户jar可以预先上传到HDFS，而不需要单独发送到集群，这就使得作业提交更加轻量了。
### **3.6 K8S 运行模式（了解）**
容器化部署是如今业界流行的一项技术，基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是Kubernetes（k8s），而Flink也在最近的版本中支持了k8s部署模式。基本原理与YARN是类似的，具体配置可以参见官网说明，这里我们就不做过多讲解了。
### **3.7 历史服务器**
运行 Flink job 的集群一旦停止，只能去 yarn 或本地磁盘上查看日志，不再可以查看作业挂掉之前的运行的 Web UI，很难清楚知道作业在挂的那一刻到底发生了什么。如果我们还没有 Metrics 监控的话，那么完全就只能通过日志去分析和定位问题了，所以如果能还原之前的 Web UI，我们可以通过 UI 发现和定位一些问题。
Flink提供了历史服务器，用来在相应的 Flink 集群关闭后查询已完成作业的统计信息。我们都知道只有当作业处于运行中的状态，才能够查看到相关的WebUI统计信息。通过 History Server 我们才能查询这些已完成作业的统计信息，无论是正常退出还是异常退出。
此外，它对外提供了 REST API，它接受 HTTP 请求并使用 JSON 数据进行响应。Flink 任务停止后，JobManager 会将已经完成任务的统计信息进行存档，History Server 进程则在任务停止后可以对任务统计信息进行查询。比如：最后一次的 Checkpoint、任务运行时的相关配置。
**1）创建存储目录**
```shell
hadoop fs -mkdir -p /logs/flink-job
```
**2）在 flink-config.yaml 中添加如下配置**
```shell
jobmanager.archive.fs.dir: hdfs://hadoop102:8020/logs/flink-job
historyserver.web.address: hadoop102
historyserver.web.port: 8082
historyserver.archive.fs.dir: hdfs://hadoop102:8020/logs/flink-job
historyserver.archive.fs.refresh-interval: 5000
```
**3）启动历史服务器**
```shell
bin/historyserver.sh start
```
**4）停止历史服务器**
```shell
bin/historyserver.sh stop
```
**5）在浏览器地址栏输入：**[**http://hadoop102:8082**](http://hadoop102:8082) **查看已经停止的** **job** **的统计信息**
## **第4章 Flink运行时架构**
### **4.1 系统架构**
**1）作业管理器（JobManager）**
JobManager是一个Flink集群中任务管理和调度的核心，是控制应用执行的主进程。也就是说，每个应用都应该被唯一的JobManager所控制执行。
JobManger又包含3个不同的组件。
**（1）JobMaster**
JobMaster是JobManager中最核心的组件，==负责处理单独的作业（Job）==。所以JobMaster和具体的Job是一一对应的，多个Job可以同时运行在一个Flink集群中, 每个Job都有一个自己的JobMaster。需要注意在早期版本的Flink中，没有JobMaster的概念；而JobManager的概念范围较小，实际指的就是现在所说的JobMaster。
在作业提交时，JobMaster会先接收到要执行的应用。JobMaster会把JobGraph转换成一个物理层面的数据流图，这个图被叫作“执行图”（ExecutionGraph），它包含了所有可以并发执行的任务。JobMaster会向资源管理器（ResourceManager）发出请求，申请执行任务必要的资源。一旦它获取到了足够的资源，就会将执行图分发到真正运行它们的TaskManager上。
而在运行过程中，JobMaster会负责所有需要中央协调的操作，比如说检查点（checkpoints）的协调。
**（2）资源管理器（ResourceManager）**
ResourceManager主要==负责资源的分配和管理==，在Flink 集群中只有一个。所谓“资源”，主要是指TaskManager的任务槽（task slots）。任务槽就是Flink集群中的资源调配单元，包含了机器用来执行计算的一组CPU和内存资源。每一个任务（Task）都需要分配到一个slot上执行。
这里注意要把Flink内置的ResourceManager和其他资源管理平台（比如YARN）的ResourceManager区分开。
**（3）分发器（Dispatcher）**
Dispatcher主要负责提供一个REST接口，==用来提交应用，并且负责为每一个新提交的作业启动一个新的JobMaster 组件==。Dispatcher也会启动一个Web UI，用来方便地展示和监控作业执行的信息。Dispatcher在架构中并不是必需的，在不同的部署模式下可能会被忽略掉。
**2）任务管理器（TaskManager）**
TaskManager是Flink中的工作进程，数据流的具体计算就是它来做的。Flink集群中必须至少有一个TaskManager；每一个TaskManager都包含了一定数量的任务槽（task slots）。Slot是资源调度的最小单位，slot的数量限制了TaskManager能够并行处理的任务数量。
启动之后，TaskManager会向资源管理器注册它的slots；收到资源管理器的指令后，TaskManager就会将一个或者多个槽位提供给JobMaster调用，JobMaster就可以分配任务来执行了。
在执行过程中，TaskManager可以缓冲数据，还可以跟其他运行同一应用的TaskManager交换数据。
### **4.2 核心概念**
#### **4.2.1 并行度（Parallelism）**
**1）并行子任务和并行度**
当要处理的数据量非常大时，我们可以把一个算子操作，“复制”多份到多个节点，数据来了之后就可以到其中任意一个执行。这样一来，一个算子任务就被拆分成了多个并行的“子任务”（subtasks），再将它们分发到不同节点，就真正实现了并行计算。
在Flink执行过程中，每一个算子（operator）可以包含一个或多个子任务（operator subtask），这些子任务在不同的线程、不同的物理机或不同的容器中完全独立地执行。
一个特定算子的==子任务（subtask）的个数==被称之为其==并行度（parallelism）==。这样，包含并行子任务的数据流，就是==并行数据流==，它需要多个分区（stream partition）来分配并行任务。一般情况下，==一个流程序的并行度==，可以认为就是==其所有算子中最大的并行度==。一个程序中，不同的算子可能具有不同的并行度。
例如：如上图所示，当前数据流中有source、map、window、sink四个算子，其中sink算子的并行度为1，其他算子的并行度都为2。所以这段流处理程序的并行度就是2。
**2）并行度的设置**
在Flink中，可以用不同的方法来设置并行度，它们的有效范围和优先级别也是不同的。
**（1）代码中设置**
我们在代码中，可以很简单地在算子后跟着调用**setParallelism()**方法，来设置当前算子的并行度：
```java
stream.map(word -> Tuple2.of(word, 1L)).setParallelism(2);
```
这种方式设置的并行度，只针对当前算子有效。
另外，我们也可以直接调用执行环境的setParallelism()方法，全局设定并行度：
```java
env.setParallelism(2);
```
这样代码中所有算子，默认的并行度就都为2了。我们一般不会在程序中设置全局并行度，因为如果在程序中对全局并行度进行硬编码，会导致无法动态扩容。
这里要注意的是，由于keyBy不是算子，所以无法对keyBy设置并行度。
**（2）提交应用时设置**
在使用flink run命令提交应用时，可以增加-p参数来指定当前应用程序执行的并行度，它的作用类似于执行环境的全局设置：
bin/flink run –p 2 –c com.atguigu.wc.SocketStreamWordCount ./FlinkTutorial-1.0-SNAPSHOT.jar
如果我们直接在Web UI上提交作业，也可以在对应输入框中直接添加并行度。
**（3）配置文件中设置**
我们还可以直接在集群的配置文件==flink-conf.yaml==中直接更改默认并行度：
```shell
parallelism.default: 2
```
这个设置对于整个集群上提交的所有作业有效，==初始值为1==。无论在代码中设置、还是提交时的-p参数，都不是必须的；所以在没有指定并行度的时候，就会采用配置文件中的集群默认并行度。在开发环境中，没有配置文件，==默认并行度就是当前机器的CPU核心数==。
#### **4.2.2 算子链（Operator Chain）**
**1）算子间的数据传输**
一个数据流在算子之间传输数据的形式可以是一对一（one-to-one）的直通（forwarding）模式，也可以是打乱的重分区（redistributing）模式，具体是哪一种形式，取决于算子的种类。
**（1）一对一（One-to-one，forwarding）**
这种模式下，数据流维护着分区以及元素的顺序。比如图中的source和map算子，source算子读取数据之后，可以直接发送给map算子做处理，它们之间==不需要重新分区，也不需要调整数据的顺序==。这就意味着map 算子的子任务，看到的元素个数和顺序跟source 算子的子任务产生的完全一样，保证着“一对一”的关系。==map、filter、flatMap等算子都是这种one-to-one的对应关系。这种关系类似于Spark中的窄依赖。==
**（2）重分区（Redistributing）**
在这种模式下，数据流的分区会发生改变。比如图中的map和后面的keyBy/window算子之间，以及keyBy/window算子和Sink算子之间，都是这样的关系。
每一个算子的子任务，会==根据数据传输的策略，把数据发送到不同的下游目标任务==。这些传输方式都会引起重分区的过程，==这一过程类似于Spark中的shuffle==。
**2）合并算子链**
在Flink中，==并行度相同的一对一（one to one）算子操作，可以直接链接在一起形成一个“大”的任务（task）==，这样原来的算子就成为了真正任务里的一部分，如下图所示。每个task会被一个线程执行。这样的技术被称为==“算子链”（Operator Chain）==。
上图中Source和map之间满足了算子链的要求，所以可以直接合并在一起，形成了一个任务；因为并行度为2，所以合并后的任务也有两个并行子任务。这样，这个数据流图所表示的作业最终会有5个任务，由5个线程并行执行。
将算子链接成task是非常有效的优化：可以减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。
Flink默认会按照算子链的原则进行链接合并，如果我们想要禁止合并或者自行定义，也可以在代码中对算子做一些特定的设置：
```java
// 禁用算子链
.map(word -> Tuple2.of(word, 1L)).disableChaining();
// 从当前算子开始新链
.map(word -> Tuple2.of(word, 1L)).startNewChain()
```
#### **4.2.3 任务槽（Task Slots）**
**1）任务槽（Task Slots）**
Flink中每一个TaskManager都是一个JVM进程，它可以启动多个独立的线程，来并行执行多个子任务（subtask）。
很显然，TaskManager的计算资源是有限的，并行的任务越多，每个线程的资源就会越少。那一个TaskManager到底能并行处理多少个任务呢？为了控制并发量，我们需要在TaskManager上对每个任务运行所占用的资源做出明确的划分，这就是所谓的任务槽（task slots）。
每个任务槽（task slot）其实表示了TaskManager拥有计算资源的一个固定大小的子集。这些资源就是用来独立执行一个子任务的。
**2）任务槽数量的设置**
在Flink的/opt/module/flink-1.17.0/conf/flink-conf.yaml配置文件中，可以设置TaskManager的slot数量，默认是1个slot。
```shell
taskmanager.numberOfTaskSlots: 8
```
需要注意的是，==slot目前仅仅用来隔离内存，不会涉及CPU的隔离==。在具体应用时，可以==将slot数量配置为机器的CPU核心数==，尽量避免不同任务之间对CPU的竞争。这也是开发环境默认并行度设为机器CPU数量的原因。
**3）任务对任务槽的共享**
默认情况下，Flink是允许子任务共享slot的。如果我们保持sink任务并行度为1不变，而作业提交时设置全局并行度为6，那么前两个任务节点就会各自有6个并行子任务，整个流处理程序则有13个子任务。如上图所示，==只要属于同一个作业==，那么对于==不同任务节点（算子）的并行子任务，就可以放到同一个slot上执行==。所以对于第一个任务节点source→map，它的6个并行子任务必须分到不同的slot上，而第二个任务节点keyBy/window/apply的并行子任务却可以和第一个任务节点共享slot。
当我们将资源密集型和非密集型的任务同时放到一个slot中，它们就可以自行分配对资源占用的比例，从而保证最重的活平均分配给所有的TaskManager。
slot共享另一个好处就是允许我们保存完整的作业管道。这样一来，即使某个TaskManager出现故障宕机，其他节点也可以完全不受影响，作业的任务可以继续执行。
当然，Flink默认是允许slot共享的，如果希望某个算子对应的任务完全独占一个slot，或者只有某一部分算子共享slot，我们也可以通过设置“slot共享组”手动指定：
```java
.map(word -> Tuple2.of(word, 1L)).slotSharingGroup("1");
```
这样，只有属于同一个slot共享组的子任务，才会开启slot共享；==不同组之间的任务是完全隔离的，必须分配到不同的slot上==。在这种场景下，总共需要的slot数量，就是各个slot共享组最大并行度的总和。
#### **4.2.4** **任务槽和并行度的关系**
==任务槽和并行度都跟程序的并行执行有关，但两者是完全不同的概念。==简单来说==任务槽是静态的概念==，是指TaskManager具有的并发执行能力，可以通过参数taskmanager.numberOfTaskSlots进行配置；而并行度是动态概念，也就是TaskManager运行程序时实际使用的并发能力，可以通过参数parallelism.default进行配置。
**举例说明：**假设一共有3个TaskManager，每一个TaskManager中的slot数量设置为3个，那么一共有9个task slot，表示集群最多能并行执行9个同一算子的子任务。
而我们定义word count程序的处理操作是四个转换算子：
source→ flatmap→ reduce→ sink
==当所有算子并行度相同时==，容易看出source和flatmap==可以合并算子链==，于是最终有三个任务节点。
通过这个例子也可以明确地看到，==整个流处理程序的并行度，就应该是所有算子并行度中最大的那个，这代表了运行程序需要的slot数量。==
### **4.3 作业提交流程**
#### **4.3.1** **Standalone会话模式作业提交流程**
#### **4.3.2 逻辑流图/作业图/执行图/物理流图**
我们已经彻底了解了由代码生成任务的过程，现在来做个梳理总结。
逻辑流图（StreamGraph）→ 作业图（JobGraph）→ 执行图（ExecutionGraph）→ 物理图（Physical Graph）。
**1）逻辑流图（StreamGraph）**
这是根据用户通过 DataStream API编写的代码生成的最初的DAG图，用来表示程序的拓扑结构。这一步一般在客户端完成。
**2）作业图（JobGraph）**
StreamGraph经过优化后生成的就是作业图（JobGraph），这是提交给 JobManager 的数据结构，确定了当前作业中所有任务的划分。主要的优化为：将多个符合条件的节点链接在一起合并成一个任务节点，形成算子链，这样可以减少数据交换的消耗。JobGraph一般也是在客户端生成的，在作业提交时传递给JobMaster。
我们提交作业之后，打开Flink自带的Web UI，点击作业就能看到对应的作业图。
**3）执行图（ExecutionGraph）**
JobMaster收到JobGraph后，会根据它来生成执行图（ExecutionGraph）。ExecutionGraph是JobGraph的并行化版本，是调度层最核心的数据结构。与JobGraph最大的区别就是按照并行度对并行子任务进行了拆分，并明确了任务间数据传输的方式。
**4）物理图（Physical Graph）**
JobMaster生成执行图后，会将它分发给TaskManager；各个TaskManager会根据执行图部署任务，最终的物理执行过程也会形成一张“图”，一般就叫作物理图（Physical Graph）。这只是具体执行层面的图，并不是一个具体的数据结构。
物理图主要就是在执行图的基础上，进一步确定数据存放的位置和收发的具体方式。有了物理图，TaskManager就可以对传递来的数据进行处理计算了。
#### **4.3.3 Yarn应用模式作业提交流程**
## **第5章** DataStream **API**
DataStream API是Flink的核心层API。一个Flink程序，其实就是对DataStream的各种转换。具体来说，代码基本上都由以下几部分构成：
### **5.1 执行环境（Execution Environment）**
Flink程序可以在各种上下文环境中运行：我们可以在本地JVM中执行程序，也可以提交到远程集群上运行。
不同的环境，代码的提交运行的过程会有所不同。这就要求我们在提交作业执行计算时，首先必须获取当前Flink的运行环境，从而建立起与Flink框架之间的联系。
#### **5.1.1 创建执行环境**
我们要获取的执行环境，是StreamExecutionEnvironment类的对象，这是所有Flink程序的基础。在代码中创建执行环境的方式，就是调用这个类的静态方法，具体有以下三种。
**1）getExecutionEnvironment**
==最简单的方式，就是直接调用getExecutionEnvironment方法。==它会根据当前运行的上下文直接得到正确的结果：如果程序是独立运行的，就返回一个本地执行环境；如果是创建了jar包，然后从命令行调用它并提交到集群执行，那么就返回集群的执行环境。也就是说，这个方法会根据当前运行的方式，自行决定该返回什么样的运行环境。
```java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
```
这种方式，用起来简单高效，是最常用的一种创建执行环境的方式。
**2）createLocalEnvironment**
==这个方法返回一个本地执行环境。==可以在调用时传入一个参数，指定默认的并行度；如果不传入，则默认并行度就是本地的CPU核心数。
```java
StreamExecutionEnvironment localEnv = StreamExecutionEnvironment.createLocalEnvironment();
```
**3）createRemoteEnvironment**
==这个方法返回集群执行环境。==需要在调用时指定JobManager的主机名和端口号，并指定要在集群中运行的Jar包。
```java
StreamExecutionEnvironment remoteEnv = StreamExecutionEnvironment
.createRemoteEnvironment(
"host", // JobManager主机名
1234, // JobManager进程端口号
"path/to/jarFile.jar" // 提交给JobManager的JAR包
);
```
在获取到程序执行环境后，我们还可以对执行环境进行灵活的设置。比如可以全局设置程序的并行度、禁用算子链，还可以定义程序的时间语义、配置容错机制。
#### **5.1.2 执行模式（Execution Mode）**
从Flink 1.12开始，官方推荐的做法是直接使用DataStream API，在提交任务时通过将执行模式设为BATCH来进行批处理。不建议使用DataSet API。
```java
// 流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
```
DataStream API执行模式包括：流执行模式、批执行模式和自动模式。
- **流执行模式（Streaming）**
这是DataStream API最经典的模式，一般用于需要持续实时处理的无界数据流。默认情况下，程序使用的就是Streaming执行模式。
- **批执行模式（Batch）**
专门用于批处理的执行模式。
- **自动模式（AutoMatic）**
在这种模式下，将由程序根据输入数据源是否有界，来自动选择执行模式。
批执行模式的使用。主要有两种方式：
**（1）通过命令行配置**
```shell
bin/flink run -Dexecution.runtime-mode=BATCH ...
```
在提交作业时，增加execution.runtime-mode参数，指定值为BATCH。
**（2）通过代码配置**
```java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
```
在代码中，直接基于执行环境调用setRuntimeMode方法，传入BATCH模式。
==实际应用中一般不会在代码中配置，而是使用命令行，这样更加灵活。==
#### **5.1.3 触发程序执行**
需要注意的是，写完输出（sink）操作并不代表程序已经结束。因为当main()方法被调用时，其实只是定义了作业的每个执行操作，然后添加到数据流图中；这时并没有真正处理数据——因为数据可能还没来。Flink是由事件驱动的，只有等到数据到来，才会触发真正的计算，这也被称为==“延迟执行”==或==“懒执行”==。
所以我们需要显式地调用执行环境的execute()方法，来触发程序执行。execute()方法将一直等待作业完成，然后返回一个执行结果（JobExecutionResult）。
env.==execute==();
### **5.2 源算子（Source）**
Flink可以从各种来源获取数据，然后构建DataStream进行转换处理。一般将数据的输入来源称为数据源（data source），而读取数据的算子就是源算子（source operator）。所以，source就是我们整个处理程序的输入端。
在Flink1.12以前，旧的添加source的方式，是调用执行环境的addSource()方法：
```java
DataStream stream = env.addSource(...);
```
方法传入的参数是一个“源函数”（source function），需要实现SourceFunction接口。
从Flink1.12开始，主要使用流批统一的新Source架构：
```java
DataStreamSource stream = env.fromSource(…)
```
Flink直接提供了很多预实现的接口，此外还有很多外部连接工具也帮我们实现了对应的Source，通常情况下足以应对我们的实际需求。
#### **5.2.1 准备工作**
为了方便练习，这里使用WaterSensor作为数据模型。
| **字段名** | **数据类型** | **说明** |
| ---------- | ------------ | ---------------- |
| **id** | String | 水位传感器类型 |
| **ts** | Long | 传感器记录时间戳 |
| **vc** | Integer | 水位记录 |
具体代码如下：
```java
public class WaterSensor {
public String id;
public Long ts;
public Integer vc;
public WaterSensor() {
}
public WaterSensor(String id, Long ts, Integer vc) {
this.id = id;
this.ts = ts;
this.vc = vc;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public Long getTs() {
return ts;
}
public void setTs(Long ts) {
this.ts = ts;
}
public Integer getVc() {
return vc;
}
public void setVc(Integer vc) {
this.vc = vc;
}
@Override
public String toString() {
return "WaterSensor{" +
"id='" + id + '\'' +
", ts=" + ts +
", vc=" + vc +
'}';
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
WaterSensor that = (WaterSensor) o;
return Objects.equals(id, that.id) &&
Objects.equals(ts, that.ts) &&
Objects.equals(vc, that.vc);
}
@Override
public int hashCode() {
return Objects.hash(id, ts, vc);
}
}
```
这里需要注意，我们定义的WaterSensor，有这样几个特点：
- 类是公有（public）的
- 有一个==无参的构造方法==
- 所有属性都是公有（public）的
- 所有属性的类型都是可以序列化的
Flink会把这样的类作为一种特殊的==POJO==（Plain Ordinary Java Object简单的Java对象，实际就是普通JavaBeans）数据类型来对待，方便数据的解析和序列化。另外我们在类中还重写了toString方法，主要是为了测试输出显示更清晰。
我们这里自定义的POJO类会在后面的代码中频繁使用，所以在后面的代码中碰到，把这里的POJO类导入就好了。
#### **5.2.2 从集合中读取数据**
最简单的读取数据的方式，就是在代码中直接创建一个Java集合，然后调用执行环境的fromCollection方法进行读取。这相当于将数据临时存储到内存中，形成特殊的数据结构后，作为数据源使用，一般用于测试。
```java
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
List data = Arrays.asList(1, 22, 3);
DataStreamSource ds = env.fromCollection(data);
stream.print();
env.execute();
}
```
#### **5.2.3 从文件读取数据**
真正的实际应用中，自然不会直接将数据写在代码中。通常情况下，我们会从存储介质中获取数据，一个比较常见的方式就是读取日志文件。这也是批处理中最常见的读取方式。
读取文件，需要添加文件连接器依赖:
```xml
org.apache.flink
flink-connector-files
${flink.version}
```
示例如下:
```java
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
FileSource fileSource = FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path("input/word.txt")).build();
env.fromSource(fileSource,WatermarkStrategy.noWatermarks(),"file")
.print();
env.execute();
}
```
说明：
- 参数可以是目录，也可以是文件；还可以从HDFS目录下读取，使用路径hdfs://...；
- 路径可以是相对路径，也可以是绝对路径；
- 相对路径是从系统属性user.dir获取路径：idea下是project的根目录，standalone模式下是集群节点根目录；
#### **5.2.4 从Socket读取数据**
不论从集合还是文件，我们读取的其实都是有界数据。在流处理的场景中，数据往往是无界的。
我们之前用到的读取socket文本流，就是流处理场景。但是这种方式由于吞吐量小、稳定性较差，一般也是用于测试。
```java
DataStream stream = env.socketTextStream("localhost", 7777);
```
#### **5.2.5 从Kafka读取数据**
Flink官方提供了连接工具flink-connector-kafka，直接帮我们实现了一个消费者FlinkKafkaConsumer，它就是用来读取Kafka数据的SourceFunction。
所以想要以Kafka作为数据源获取数据，我们只需要引入Kafka连接器的依赖。Flink官方提供的是一个通用的Kafka连接器，它会自动跟踪最新版本的Kafka客户端。目前最新版本只支持0.10.0版本以上的Kafka。这里我们需要导入的依赖如下。
```xml
org.apache.flink
flink-connector-kafka
${flink.version}
```
代码如下：
```java
public class SourceKafka {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
KafkaSource kafkaSource = KafkaSource.builder()
.setBootstrapServers("hadoop102:9092")
.setTopics("topic_1")
.setGroupId("atguigu")
.setStartingOffsets(OffsetsInitializer.latest())
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
DataStreamSource stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source");
stream.print("Kafka");
env.execute();
}
}
```
#### **5.2.6 从数据生成器读取数据**
Flink从1.11开始提供了一个内置的DataGen 连接器，主要是用于生成一些随机数，用于在没有数据源的时候，进行流任务的测试以及性能测试等。1.17提供了新的Source写法，需要导入依赖：
```xml
org.apache.flink
flink-connector-datagen
${flink.version}
```
代码如下：
```java
public class DataGeneratorDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataGeneratorSource dataGeneratorSource =
new DataGeneratorSource<>(
new GeneratorFunction() {
@Override
public String map(Long value) throws Exception {
return "Number:"+value;
}
},
Long.MAX_VALUE,
RateLimiterStrategy.perSecond(10),
Types.STRING
);
env
.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "datagenerator")
.print();
env.execute();
}
}
```
#### **5.2.7 Flink支持的数据类型**
**1）Flink的类型系统**
Flink使用“类型信息”（TypeInformation）来统一表示数据类型。TypeInformation类是Flink中所有类型描述符的基类。它涵盖了类型的一些基本属性，并为每个数据类型生成特定的序列化器、反序列化器和比较器。
**2）Flink支持的数据类型**
对于常见的Java和Scala数据类型，Flink都是支持的。Flink在内部，Flink对支持不同的类型进行了划分，这些类型可以在Types工具类中找到：
==（1）基本类型==
所有Java基本类型及其包装类，再加上Void、String、Date、BigDecimal和BigInteger。
==（2）数组类型==
包括基本类型数组（PRIMITIVE_ARRAY）和对象数组（OBJECT_ARRAY）。
==（3）复合数据类型==
- Java元组类型（TUPLE）：这是Flink内置的元组类型，是Java API的一部分。最多25个字段，也就是从Tuple0~Tuple25，不支持空字段。
- Scala 样例类及Scala元组：不支持空字段。
- 行类型（ROW）：可以认为是具有任意个字段的元组，并支持空字段。
- POJO：Flink自定义的类似于Java bean模式的类。
==（4）辅助类型==
Option、Either、List、Map等。
==（5）泛型类型（GENERIC）==
Flink支持所有的Java类和Scala类。不过如果没有按照上面POJO类型的要求来定义，就会被Flink当作泛型类来处理。Flink会把泛型类型当作黑盒，无法获取它们内部的属性；它们也不是由Flink本身序列化的，==而是由Kryo序列化的==。
在这些类型中，元组类型和POJO类型最为灵活，因为它们支持创建复杂类型。而相比之下，POJO还支持在键（key）的定义中直接使用字段名，这会让我们的代码可读性大大增加。所以，在项目实践中，往往会将流处理程序中的元素类型定为Flink的POJO类型。
Flink对POJO类型的要求如下：
- 类是公有（public）的
- 有一个无参的构造方法
- 所有属性都是公有（public）的
- 所有属性的类型都是可以序列化的
**3）类型提示（Type Hints）**
Flink还具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息，从而获得对应的序列化器和反序列化器。但是，由于Java中泛型擦除的存在，在某些特殊情况下（比如Lambda表达式中），自动提取的信息是不够精细的——只告诉Flink当前的元素由“船头、船身、船尾”构成，根本无法重建出“大船”的模样；这时就需要显式地提供类型信息，才能使应用程序正常工作或提高其性能。
为了解决这类问题，Java API提供了专门的“类型提示”（type hints）。
回忆一下之前的word count流处理程序，我们在将String类型的每个词转换成（word， count）二元组后，就明确地用returns指定了返回的类型。因为对于map里传入的Lambda表达式，系统只能推断出返回的是Tuple2类型，而无法得到Tuple2。只有显式地告诉系统当前的返回类型，才能正确地解析出完整数据。
```java
.map(word -> Tuple2.of(word, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));
```
Flink还专门提供了TypeHint类，它可以捕获泛型的类型信息，并且一直记录下来，为运行时提供足够的信息。我们同样可以通过.returns()方法，明确地指定转换之后的DataStream里元素的类型。
```java
returns(new TypeHint>(){})
```
### **5.3 转换算子（Transformation）**
数据源读入数据之后，我们就可以使用各种转换算子，将一个或多个DataStream转换为新的DataStream。
#### **5.3.1 基本转换算子（map/filter/flat/Map）**
##### **5.3.1.1** **映射（map）**
map是大家非常熟悉的大数据操作算子，主要用于将数据流中的数据进行转换，形成新的数据流。简单来说，就是一个“一一映射”，消费一个元素就产出一个元素。
我们只需要基于DataStream调用map()方法就可以进行转换处理。方法需要传入的参数是接口MapFunction的实现；返回值类型还是DataStream，不过泛型（流中的元素类型）可能改变。
下面的代码用不同的方式，实现了提取WaterSensor中的id字段的功能。
```java
public class TransMap {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource stream = env.fromElements(
new WaterSensor("sensor_1", 1, 1),
new WaterSensor("sensor_2", 2, 2)
);
// 方式一：传入匿名类，实现MapFunction
stream.map(new MapFunction() {
@Override
public String map(WaterSensor e) throws Exception {
return e.id;
}
}).print();
// 方式二：传入MapFunction的实现类
// stream.map(new UserMap()).print();
env.execute();
}
public static class UserMap implements MapFunction {
@Override
public String map(WaterSensor e) throws Exception {
return e.id;
}
}
}
```
上面代码中，MapFunction实现类的泛型类型，与输入数据类型和输出数据的类型有关。在实现MapFunction接口的时候，需要指定两个泛型，分别是输入事件和输出事件的类型，还需要重写一个map()方法，定义从一个输入事件转换为另一个输出事件的具体逻辑。
##### **5.3.1.2** **过滤（filter）**
filter转换操作，顾名思义是对数据流执行一个过滤，通过一个布尔条件表达式设置过滤条件，对于每一个流内元素进行判断，若为true则元素正常输出，若为false则元素被过滤掉。
进行filter转换之后的新数据流的数据类型与原数据流是相同的。filter转换需要传入的参数需要实现FilterFunction接口，而FilterFunction内要实现filter()方法，就相当于一个返回布尔类型的条件表达式。
**案例需求：**下面的代码会将数据流中传感器id为sensor_1的数据过滤出来。
```java
public class TransFilter {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource stream = env.fromElements(
new WaterSensor("sensor_1", 1, 1),
new WaterSensor("sensor_1", 2, 2),
new WaterSensor("sensor_2", 2, 2),
new WaterSensor("sensor_3", 3, 3)
);
// 方式一：传入匿名类实现FilterFunction
stream.filter(new FilterFunction() {
@Override
public boolean filter(WaterSensor e) throws Exception {
return e.id.equals("sensor_1");
}
}).print();
// 方式二：传入FilterFunction实现类
// stream.filter(new UserFilter()).print();
env.execute();
}
public static class UserFilter implements FilterFunction {
@Override
public boolean filter(WaterSensor e) throws Exception {
return e.id.equals("sensor_1");
}
}
}
```
##### **5.3.1.3 扁平映射（flatMap）**
flatMap操作又称为扁平映射，主要是将数据流中的整体（一般是集合类型）拆分成一个一个的个体使用。消费一个元素，可以产生0到多个元素。flatMap可以认为是“扁平化”（flatten）和“映射”（map）两步操作的结合，也就是先按照某种规则对数据进行打散拆分，再对拆分后的元素做转换处理。
同map一样，flatMap也可以使用Lambda表达式或者FlatMapFunction接口实现类的方式来进行传参，返回值类型取决于所传参数的具体逻辑，可以与原数据流相同，也可以不同。
**案例需求：**如果输入的数据是sensor_1，只打印vc；如果输入的数据是sensor_2，既打印ts又打印vc。
实现代码如下：
```java
public class TransFlatmap {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource stream = env.fromElements(
new WaterSensor("sensor_1", 1, 1),
new WaterSensor("sensor_1", 2, 2),
new WaterSensor("sensor_2", 2, 2),
new WaterSensor("sensor_3", 3, 3)
);
stream.flatMap(new MyFlatMap()).print();
env.execute();
}
public static class MyFlatMap implements FlatMapFunction {
@Override
public void flatMap(WaterSensor value, Collector out) throws Exception {
if (value.id.equals("sensor_1")) {
out.collect(String.valueOf(value.vc));
} else if (value.id.equals("sensor_2")) {
out.collect(String.valueOf(value.ts));
out.collect(String.valueOf(value.vc));
}
}
}
}
```
#### **5.3.2 聚合算子（Aggregation）**
计算的结果不仅依赖当前数据，还跟之前的数据有关，相当于要把所有数据聚在一起进行汇总合并——这就是所谓的“聚合”（Aggregation），类似于MapReduce中的reduce操作。
##### **5.3.2.1** **按键分区（keyBy）**
对于Flink而言，DataStream是没有直接进行聚合的API的。因为我们对海量数据做聚合肯定要进行分区并行处理，这样才能提高效率。所以在Flink中，要做聚合，需要先进行分区；这个操作就是通过keyBy来完成的。
keyBy是聚合前必须要用到的一个算子。keyBy通过指定键（key），可以将一条流从逻辑上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务。
基于不同的key，流中的数据将被分配到不同的分区中去；这样一来，所有具有相同的key的数据，都将被发往同一个分区。
在内部，是通过计算==key的哈希值（hash code）==，==对分区数进行取模运算来实现的==。所以这里key如果是POJO的话，必须要重写hashCode()方法。
keyBy()方法需要传入一个参数，这个参数指定了一个或一组key。有很多不同的方法来指定key：比如对于Tuple数据类型，可以指定字段的位置或者多个位置的组合；对于POJO类型，可以指定字段的名称（String）；另外，还可以传入Lambda表达式或者实现一个键选择器（KeySelector），用于说明从数据中提取key的逻辑。
我们可以以id作为key做一个分区操作，代码实现如下：
```java
public class TransKeyBy {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource stream = env.fromElements(
new WaterSensor("sensor_1", 1, 1),
new WaterSensor("sensor_1", 2, 2),
new WaterSensor("sensor_2", 2, 2),
new WaterSensor("sensor_3", 3, 3)
);
// 方式一：使用Lambda表达式
KeyedStream keyedStream = stream.keyBy(e -> e.id);
// 方式二：使用匿名类实现KeySelector
KeyedStream keyedStream1 = stream.keyBy(new KeySelector() {
@Override
public String getKey(WaterSensor e) throws Exception {
return e.id;
}
});
env.execute();
}
}
```
需要注意的是，keyBy得到的结果将不再是DataStream，而是会将DataStream转换为KeyedStream。KeyedStream可以认为是“分区流”或者“键控流”，它是对DataStream按照key的一个逻辑分区，所以泛型有两个类型：除去当前流中的元素类型外，还需要指定key的类型。
KeyedStream也继承自DataStream，所以基于它的操作也都归属于DataStream API。但它跟之前的转换操作得到的SingleOutputStreamOperator不同，只是一个流的分区操作，并不是一个转换算子。KeyedStream是一个非常重要的数据结构，只有基于它才可以做后续的聚合操作（比如sum，reduce）。
##### **5.3.2.2** **简单聚合（sum/min/max/minBy/maxBy）**
有了按键分区的数据流KeyedStream，我们就可以基于它进行聚合操作了。Flink为我们内置实现了一些最基本、最简单的聚合API，主要有以下几种：
- ==sum()==：在输入流上，对指定的字段做叠加求和的操作。
- ==min()==：在输入流上，对指定的字段求最小值。
- ==max()==：在输入流上，对指定的字段求最大值。
- ==minBy()==：与min()类似，在输入流上针对指定字段求最小值。不同的是，min()只计算指定字段的最小值，其他字段会保留最初第一个数据的值；而minBy()则会返回包含字段最小值的整条数据。
- ==maxBy()==：与max()类似，在输入流上针对指定字段求最大值。两者区别与min()/minBy()完全一致。
简单聚合算子使用非常方便，语义也非常明确。这些聚合方法调用时，也需要传入参数；但并不像基本转换算子那样需要实现自定义函数，只要说明聚合指定的字段就可以了。指定字段的方式有两种：指定位置，和指定名称。
对于元组类型的数据，可以使用这两种方式来指定字段。需要注意的是，元组中字段的名称，是以f0、f1、f2、…来命名的。
如果数据流的类型是POJO类，那么就只能通过字段名称来指定，不能通过位置来指定了。
```java
public class TransAggregation {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource stream = env.fromElements(
new WaterSensor("sensor_1", 1, 1),
new WaterSensor("sensor_1", 2, 2),
new WaterSensor("sensor_2", 2, 2),
new WaterSensor("sensor_3", 3, 3)
);
stream.keyBy(e -> e.id).max("vc"); // 指定字段名称
env.execute();
}
}
```
简单聚合算子返回的，同样是一个SingleOutputStreamOperator，也就是从KeyedStream又转换成了常规的DataStream。所以可以这样理解：keyBy和聚合是成对出现的，先分区、后聚合，得到的依然是一个DataStream。而且经过简单聚合之后的数据流，元素的数据类型保持不变。
一个聚合算子，会为每一个key保存一个聚合的值，在Flink中我们把它叫作“状态”（state）。所以每当有一个新的数据输入，算子就会更新保存的聚合结果，并发送一个带有更新后聚合值的事件到下游算子。对于无界流来说，这些状态是永远不会被清除的，所以我们使用聚合算子，应该只用在含有有限个key的数据流上。
##### **5.3.2.3** **归约聚合（reduce）**
reduce可以对已有的数据进行归约处理，把每一个新输入的数据和当前已经归约出来的值，再做一个聚合计算。
reduce操作也会将KeyedStream转换为DataStream。它不会改变流的元素数据类型，所以输出类型和输入类型是一样的。
调用KeyedStream的reduce方法时，需要传入一个参数，实现ReduceFunction接口。接口在源码中的定义如下：
```java
public interface ReduceFunction extends Function, Serializable {
T reduce(T value1, T value2) throws Exception;
}
```
ReduceFunction接口里需要实现reduce()方法，这个方法接收两个输入事件，经过转换处理之后输出一个相同类型的事件。在流处理的底层实现过程中，实际上是将中间“合并的结果”作为任务的一个状态保存起来的；之后每来一个新的数据，就和之前的聚合状态进一步做归约。
我们可以单独定义一个函数类实现ReduceFunction接口，==也可以直接传入一个匿名类==。当然，同样也可以==通过传入Lambda表达式实现类似的功能==。
为了方便后续使用，定义一个WaterSensorMapFunction：
```java
public class WaterSensorMapFunction implements MapFunction {
@Override
public WaterSensor map(String value) throws Exception {
String[] datas = value.split(",");
return new WaterSensor(datas[0],Long.valueOf(datas[1]) ,Integer.valueOf(datas[2]) );
}
}
```
案例：使用reduce实现max和maxBy的功能。
```java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction())
.keyBy(WaterSensor::getId)
.reduce(new ReduceFunction()
{
@Override
public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
System.out.println("Demo7_Reduce.reduce");
int maxVc = Math.max(value1.getVc(), value2.getVc());
//实现max(vc)的效果 取最大值，其他字段以当前组的第一个为主
//value1.setVc(maxVc);
//实现maxBy(vc)的效果 取当前最大值的所有字段
if (value1.getVc() > value2.getVc()){
value1.setVc(maxVc);
return value1;
}else {
value2.setVc(maxVc);
return value2;
}
}
})
.print();
env.execute();
```
reduce同简单聚合算子一样，也要针对每一个key保存状态。因为状态不会清空，所以我们需要将reduce算子作用在一个有限key的流上。
#### **5.3.3 用户自定义函数（UDF）**
用户自定义函数（user-defined function，UDF），即用户可以根据自身需求，重新实现算子的逻辑。
用户自定义函数分为：函数类、匿名函数、富函数类。
##### **5.3.3.1 函数类（Function Classes）**
Flink暴露了所有UDF函数的接口，具体实现方式为接口或者抽象类，例如MapFunction、FilterFunction、ReduceFunction等。所以用户可以自定义一个函数类，实现对应的接口。
**需求：**用来从用户的点击数据中筛选包含“sensor_1”的内容：
**方式一：**实现FilterFunction接口
```java
public class TransFunctionUDF {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource stream = env.fromElements(
new WaterSensor("sensor_1", 1, 1),
new WaterSensor("sensor_1", 2, 2),
new WaterSensor("sensor_2", 2, 2),
new WaterSensor("sensor_3", 3, 3)
);
DataStream filter = stream.filter(new UserFilter());
filter.print();
env.execute();
}
public static class UserFilter implements FilterFunction {
@Override
public boolean filter(WaterSensor e) throws Exception {
return e.id.equals("sensor_1");
}
}
}
```
**方式二：**通过匿名类来实现FilterFunction接口：
```java
DataStream stream = stream.filter(new FilterFunction< WaterSensor>() {
@Override
public boolean filter(WaterSensor e) throws Exception {
return e.id.equals("sensor_1");
}
});
```
**方式二的优化：**为了类可以更加通用，我们还可以将用于过滤的关键字"home"抽象出来作为类的属性，调用构造方法时传进去。
```java
DataStreamSource stream = env.fromElements(
new WaterSensor("sensor_1", 1, 1),
new WaterSensor("sensor_1", 2, 2),
new WaterSensor("sensor_2", 2, 2),
new WaterSensor("sensor_3", 3, 3)
);
DataStream stream = stream.filter(new FilterFunctionImpl("sensor_1"));
public static class FilterFunctionImpl implements FilterFunction {
private String id;
FilterFunctionImpl(String id) { this.id=id; }
@Override
public boolean filter(WaterSensor value) throws Exception {
return thid.id.equals(value.id);
}
}
```
**方式三：**采用匿名函数（Lambda）
```java
public class TransFunctionUDF {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource stream = env.fromElements(
new WaterSensor("sensor_1", 1, 1),
new WaterSensor("sensor_1", 2, 2),
new WaterSensor("sensor_2", 2, 2),
new WaterSensor("sensor_3", 3, 3)
);
//map函数使用Lambda表达式，不需要进行类型声明
SingleOutputStreamOperator filter = stream.filter(sensor -> "sensor_1".equals(sensor.id));
filter.print();
env.execute();
}
}
```
##### **5.3.3.2** **富函数类（Rich Function Classes）**
“富函数类”也是DataStream API提供的一个函数类的接口，所有的Flink函数类都有其Rich版本。富函数类一般是以抽象类的形式出现的。例如：RichMapFunction、RichFilterFunction、RichReduceFunction等。
与常规函数类的不同主要在于，富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。
Rich Function有生命周期的概念。典型的生命周期方法有：
- ==open()方法==，是Rich Function的初始化方法，也就是会开启一个算子的生命周期。当一个算子的实际工作方法例如map()或者filter()方法被调用之前，open()会首先被调用。
- ==close()方法==，是生命周期中的最后一个调用的方法，类似于结束方法。一般用来做一些清理工作。
需要注意的是，这里的生命周期方法，对于一个并行子任务来说只会调用一次；而对应的，实际工作方法，例如RichMapFunction中的map()，在每条数据到来后都会触发一次调用。
来看一个例子说明：
```java
public class RichFunctionExample {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
env
.fromElements(1,2,3,4)
.map(new RichMapFunction() {
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期开始");
}
@Override
public Integer map(Integer integer) throws Exception {
return integer + 1;
}
@Override
public void close() throws Exception {
super.close();
System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期结束");
}
})
.print();
env.execute();
}
}
```
#### **5.3.4 物理分区算子（Physical Partitioning）**
常见的物理分区策略有：随机分配（Random）、轮询分配（Round-Robin）、重缩放（Rescale）和广播（Broadcast）。
##### **5.3.4.1** **随机分区（shuffle）**
最简单的重分区方式就是直接“洗牌”。通过调用DataStream的.shuffle()方法，将数据随机地分配到下游算子的并行任务中去。
随机分区服从均匀分布（uniform distribution），所以可以把流中的数据随机打乱，均匀地传递到下游任务分区。因为是完全随机的，所以对于同样的输入数据, 每次执行得到的结果也不会相同。
经过随机分区之后，得到的依然是一个DataStream。
我们可以做个简单测试：将数据读入之后直接打印到控制台，将输出的并行度设置为2，中间经历一次shuffle。执行多次，观察结果是否相同。
```java
public class ShuffleExample {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
DataStreamSource stream = env.socketTextStream("hadoop102", 7777);;
stream.shuffle().print()
env.execute();
}
}
```
##### **5.3.4.2 轮询分区（Round-Robin）**
轮询，简单来说就是“发牌”，按照先后顺序将数据做依次分发。通过调用DataStream的.rebalance()方法，就可以实现轮询重分区。rebalance使用的是Round-Robin负载均衡算法，可以将输入流数据平均分配到下游的并行任务中去。
##### **5.3.4.3** **重缩放分区（rescale）**
重缩放分区和轮询分区非常相似。当调用rescale()方法时，其实底层也是使用Round-Robin算法进行轮询，但是只会将数据==轮询发送到下游并行任务的一部分中==。rescale的做法是分成小团体，发牌人==只给自己团体内的所有人轮流发牌==。
```java
stream.rescale()
```
**5.3.4.5 全局分区（global）**
全局分区也是一种特殊的分区方式。这种做法非常极端，通过调用.global()方法，会将所有的输入流数据都发送到下游算子的第一个并行子任务中去。这就相当于强行让下游任务并行度变成了1，所以使用这个操作需要非常谨慎，可能对程序造成很大的压力。
```java
stream.global()
```
##### **5.3.4.6** **自定义分区（Custom）**
当Flink提供的所有分区策略都不能满足用户的需求时，我们可以通过使用partitionCustom()方法来自定义分区策略。
**1）自定义分区器**
```java
public class MyPartitioner implements Partitioner {
@Override
public int partition(String key, int numPartitions) {
return Integer.parseInt(key) % numPartitions;
}
}
```
**2）使用自定义分区**
```java
public class PartitionCustomDemo {
public static void main(String[] args) throws Exception {
// StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
env.setParallelism(2);
DataStreamSource socketDS = env.socketTextStream("hadoop102", 7777);
DataStream myDS = socketDS
.partitionCustom(
new MyPartitioner(),
value -> value);
myDS.print();
env.execute();
}
}
```
#### **5.3.5** **分流**
所谓“分流”，就是将一条数据流拆分成完全独立的两条、甚至多条流。也就是基于一个DataStream，定义一些筛选条件，将符合条件的数据拣选出来放到对应的流里。
##### **5.3.5.1** **简单实现**
其实根据条件筛选数据的需求，本身非常容易实现：只要针对同一条流多次独立调用.filter()方法进行筛选，就可以得到拆分之后的流了。
**案例需求：**读取一个整数数字流，将数据流划分为奇数流和偶数流。
**代码实现：**
```java
public class SplitStreamByFilter {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
SingleOutputStreamOperator ds = env.socketTextStream("hadoop102", 7777)
.map(Integer::valueOf);
//将ds 分为两个流 ，一个是奇数流，一个是偶数流
//使用filter 过滤两次
SingleOutputStreamOperator ds1 = ds.filter(x -> x % 2 == 0);
SingleOutputStreamOperator ds2 = ds.filter(x -> x % 2 == 1);
ds1.print("偶数");
ds2.print("奇数");
env.execute();
}
}
```
这种实现非常简单，但代码显得有些冗余——我们的处理逻辑对拆分出的三条流其实是一样的，却重复写了三次。而且这段代码背后的含义，==是将原始数据流stream复制三份，然后对每一份分别做筛选==；这明显是不够高效的。我们自然想到，能不能不用复制流，直接用一个算子就把它们都拆分开呢？
##### **5.3.5.2** **使用侧输出流**
关于处理函数中侧输出流的用法，我们已经在7.5节做了详细介绍。简单来说，只需要调用上下文ctx的.output()方法，就可以输出任意类型的数据了。而侧输出流的标记和提取，都离不开一个“输出标签”（OutputTag），指定了侧输出流的id和类型。
**代码实现：**将WaterSensor按照Id类型进行分流。
```java
public class SplitStreamByOutputTag {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
SingleOutputStreamOperator ds = env.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction());
OutputTag s1 = new OutputTag<>("s1", Types.POJO(WaterSensor.class)){};
OutputTag s2 = new OutputTag<>("s2", Types.POJO(WaterSensor.class)){};
//返回的都是主流
SingleOutputStreamOperator ds1 = ds.process(new ProcessFunction()
{
@Override
public void processElement(WaterSensor value, Context ctx, Collector out) throws Exception {
if ("s1".equals(value.getId())) {
ctx.output(s1, value);
} else if ("s2".equals(value.getId())) {
ctx.output(s2, value);
} else {
//主流
out.collect(value);
}
}
});
ds1.print("主流，非s1,s2的传感器");
SideOutputDataStream s1DS = ds1.getSideOutput(s1);
SideOutputDataStream s2DS = ds1.getSideOutput(s2);
s1DS.printToErr("s1");
s2DS.printToErr("s2");
env.execute();
}
}
```
#### **5.3.6** **基本合流操作**
在实际应用中，我们经常会遇到来源不同的多条流，需要将它们的数据进行联合处理。所以Flink中合流的操作会更加普遍，对应的API也更加丰富。
##### **5.3.6.1** **联合（Union）**
最简单的合流操作，就是直接将多条流合在一起，叫作流的“联合”（union）。联合操作要求必须流中的==数据类型必须相同==，合并之后的新流会包括所有流中的元素，数据类型不变。
注意：union()的参数可以是多个DataStream，所以联合操作可以实现多条流的合并。
**代码实现：**我们可以用下面的代码做一个简单测试：
```java
public class UnionExample {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource ds1 = env.fromElements(1, 2, 3);
DataStreamSource ds2 = env.fromElements(2, 2, 3);
DataStreamSource ds3 = env.fromElements("2", "2", "3");
ds1.union(ds2,ds3.map(Integer::valueOf))
.print();
env.execute();
}
}
```
##### **5.3.6.2** **连接（Connect）**
==流的联合虽然简单，不过受限于数据类型不能改变==，灵活性大打折扣，所以实际应用较少出现。除了联合（union），Flink还提供了另外一种方便的合流操作——==连接（connect）==。
**1）连接流（ConnectedStreams）**
**代码实现：**需要分为两步：首先基于一条DataStream调用.connect()方法，传入另外一条DataStream作为参数，将两条流连接起来，得到一个ConnectedStreams；然后再调用同处理方法得到DataStream。这里可以的调用的同处理方法有.map()/.flatMap()，以及.process()方法。
```java
public class ConnectDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// DataStreamSource source1 = env.fromElements(1, 2, 3);
// DataStreamSource source2 = env.fromElements("a", "b", "c");
SingleOutputStreamOperator source1 = env
.socketTextStream("hadoop102", 7777)
.map(i -> Integer.parseInt(i));
DataStreamSource source2 = env.socketTextStream("hadoop102", 8888);
/**
* TODO 使用 connect 合流
* 1、一次只能连接 2条流
* 2、流的数据类型可以不一样
* 3、 连接后可以调用 map、flatmap、process来处理，但是各处理各的
*/
ConnectedStreams connect = source1.connect(source2);
SingleOutputStreamOperator result = connect.map(new CoMapFunction() {
@Override
public String map1(Integer value) throws Exception {
return "来源于数字流:" + value.toString();
}
@Override
public String map2(String value) throws Exception {
return "来源于字母流:" + value;
}
});
result.print();
env.execute();
}
}
```
上面的代码中，==ConnectedStreams==有两个类型参数，分别表示内部包含的两条流各自的数据类型；由于需要“一国两制”，因此调用.map()方法时传入的不再是一个简单的MapFunction，而是一个CoMapFunction，表示分别对两条流中的数据执行map操作。这个接口有三个类型参数，依次表示第一条流、第二条流，以及合并后的流中的数据类型。需要实现的方法也非常直白：==.map1()就是对第一条流中数据的map操作，.map2()则是针对第二条流。==
**2）CoProcessFunction**
与CoMapFunction类似，如果是调用.map()就需要传入一个CoMapFunction，需要实现map1()、map2()两个方法；而调用.process()时，传入的则是一个CoProcessFunction。它也是“处理函数”家族中的一员，用法非常相似。它需要实现的就是processElement1()、processElement2()两个方法，在每个数据到来时，会根据来源的流调用其中的一个方法进行处理。
值得一提的是，ConnectedStreams也可以直接调用.keyBy()进行按键分区的操作，得到的还是一个ConnectedStreams：
```java
connectedStreams.keyBy(keySelector1, keySelector2);
```
这里传入两个参数keySelector1和keySelector2，是两条流中各自的键选择器；当然也可以直接传入键的位置值（keyPosition），或者键的字段名（field），这与普通的keyBy用法完全一致。ConnectedStreams进行keyBy操作，其实就是==把两条流中key相同的数据放到了一起==，然后针对来源的流再做各自处理，这在一些场景下非常有用。
**案例需求：**连接两条流，输出能根据id匹配上的数据（类似inner join效果）
```java
public class ConnectKeybyDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
DataStreamSource> source1 = env.fromElements(
Tuple2.of(1, "a1"),
Tuple2.of(1, "a2"),
Tuple2.of(2, "b"),
Tuple2.of(3, "c")
);
DataStreamSource> source2 = env.fromElements(
Tuple3.of(1, "aa1", 1),
Tuple3.of(1, "aa2", 2),
Tuple3.of(2, "bb", 1),
Tuple3.of(3, "cc", 1)
);
ConnectedStreams, Tuple3> connect = source1.connect(source2);
// 多并行度下，需要根据 关联条件 进行keyby，才能保证key相同的数据到一起去，才能匹配上
ConnectedStreams, Tuple3> connectKey = connect.keyBy(s1 -> s1.f0, s2 -> s2.f0);
SingleOutputStreamOperator result = connectKey.process(
new CoProcessFunction, Tuple3, String>() {
// 定义 HashMap，缓存来过的数据，key=id，value=list<数据>
Map>> s1Cache = new HashMap<>();
Map>> s2Cache = new HashMap<>();
@Override
public void processElement1(Tuple2 value, Context ctx, Collector out) throws Exception {
Integer id = value.f0;
// TODO 1.来过的s1数据，都存起来
if (!s1Cache.containsKey(id)) {
// 1.1 第一条数据，初始化 value的list，放入 hashmap
List> s1Values = new ArrayList<>();
s1Values.add(value);
s1Cache.put(id, s1Values);
} else {
// 1.2 不是第一条，直接添加到 list中
s1Cache.get(id).add(value);
}
//TODO 2.根据id，查找s2的数据，只输出 匹配上 的数据
if (s2Cache.containsKey(id)) {
for (Tuple3 s2Element : s2Cache.get(id)) {
out.collect("s1:" + value + "<--------->s2:" + s2Element);
}
}
}
@Override
public void processElement2(Tuple3 value, Context ctx, Collector out) throws Exception {
Integer id = value.f0;
// TODO 1.来过的s2数据，都存起来
if (!s2Cache.containsKey(id)) {
// 1.1 第一条数据，初始化 value的list，放入 hashmap
List> s2Values = new ArrayList<>();
s2Values.add(value);
s2Cache.put(id, s2Values);
} else {
// 1.2 不是第一条，直接添加到 list中
s2Cache.get(id).add(value);
}
//TODO 2.根据id，查找s1的数据，只输出 匹配上 的数据
if (s1Cache.containsKey(id)) {
for (Tuple2 s1Element : s1Cache.get(id)) {
out.collect("s1:" + s1Element + "<--------->s2:" + value);
}
}
}
});
result.print();
env.execute();
}
}
```
### **5.4 输出算子（Sink）**
Flink作为数据处理框架，最终还是要把计算处理的结果写入外部存储，为外部应用提供支持。
#### **5.4.1 连接到外部系统**
Flink的DataStream API专门提供了向外部写入数据的方法：addSink。与addSource类似，addSink方法对应着一个“Sink”算子，主要就是用来实现与外部系统连接、并将数据提交写入的；Flink程序中所有对外的输出操作，一般都是利用Sink算子完成的。
Flink1.12以前，Sink算子的创建是通过调用DataStream的.addSink()方法实现的。
```
stream.addSink(new SinkFunction(…));
```
addSink方法同样需要传入一个参数，实现的是SinkFunction接口。在这个接口中只需要重写一个方法invoke()，用来将指定的值写入到外部系统中。这个方法在每条数据记录到来时都会调用。
Flink1.12开始，同样重构了Sink架构，
```
stream.sinkTo(…)
```
当然，Sink多数情况下同样并不需要我们自己实现。之前我们一直在使用的print方法其实就是一种Sink，它表示将数据流写入标准控制台打印输出。Flink官方为我们提供了一部分的框架的Sink连接器。如下图所示，列出了Flink官方目前支持的第三方系统连接器：
我们可以看到，像Kafka之类流式系统，Flink提供了完美对接，source/sink两端都能连接，可读可写；而对于Elasticsearch、JDBC等数据存储系统，则只提供了输出写入的sink连接器。
除Flink官方之外，Apache Bahir框架，也实现了一些其他第三方系统与Flink的连接器。
除此以外，就需要用户自定义实现sink连接器了。
#### **5.4.2 输出到文件**
Flink专门提供了一个流式文件系统的连接器：FileSink，为批处理和流处理提供了一个统一的Sink，它可以将分区文件写入Flink支持的文件系统。
FileSink支持行编码（Row-encoded）和批量编码（Bulk-encoded）格式。这两种不同的方式都有各自的构建器（builder），可以直接调用FileSink的静态方法：
- 行编码： FileSink.forRowFormat（basePath，rowEncoder）。
- 批量编码： FileSink.forBulkFormat（basePath，bulkWriterFactory）。
示例:
```java
public class SinkFile {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 每个目录中，都有 并行度个数的 文件在写入
env.setParallelism(2);
// 必须开启checkpoint，否则一直都是 .inprogress
env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
DataGeneratorSource dataGeneratorSource = new DataGeneratorSource<>(
new GeneratorFunction() {
@Override
public String map(Long value) throws Exception {
return "Number:" + value;
}
},
Long.MAX_VALUE,
RateLimiterStrategy.perSecond(1000),
Types.STRING
);
DataStreamSource dataGen = env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "data-generator");
// 输出到文件系统
FileSink fieSink = FileSink
// 输出行式存储的文件，指定路径、指定编码
.forRowFormat(new Path("f:/tmp"), new SimpleStringEncoder<>("UTF-8"))
// 输出文件的一些配置： 文件名的前缀、后缀
.withOutputFileConfig(
OutputFileConfig.builder()
.withPartPrefix("atguigu-")
.withPartSuffix(".log")
.build()
)
// 按照目录分桶：如下，就是每个小时一个目录
.withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd HH", ZoneId.systemDefault()))
// 文件滚动策略: 1分钟 或 1m
.withRollingPolicy(
DefaultRollingPolicy.builder()
.withRolloverInterval(Duration.ofMinutes(1))
.withMaxPartSize(new MemorySize(1024*1024))
.build()
)
.build();
dataGen.sinkTo(fieSink);
env.execute();
}
}
```
#### **5.4.3 输出到Kafka**
（1）添加Kafka 连接器依赖
由于我们已经测试过从Kafka数据源读取数据，连接器相关依赖已经引入，这里就不重复介绍了。
（2）启动Kafka集群
（3）编写输出到Kafka的示例代码
输出无key的record:
```java
public class SinkKafka {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 如果是精准一次，必须开启checkpoint（后续章节介绍）
env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
SingleOutputStreamOperator sensorDS = env
.socketTextStream("hadoop102", 7777);
/**
* Kafka Sink:
* TODO 注意：如果要使用 精准一次 写入Kafka，需要满足以下条件，缺一不可
* 1、开启checkpoint（后续介绍）
* 2、设置事务前缀
* 3、设置事务超时时间： checkpoint间隔 < 事务超时时间 < max的15分钟
*/
KafkaSink kafkaSink = KafkaSink.builder()
// 指定 kafka 的地址和端口
.setBootstrapServers("hadoop102:9092,hadoop103:9092,hadoop104:9092")
// 指定序列化器：指定Topic名称、具体的序列化
.setRecordSerializer(
KafkaRecordSerializationSchema.builder()
.setTopic("ws")
.setValueSerializationSchema(new SimpleStringSchema())
.build()
)
// 写到kafka的一致性级别： 精准一次、至少一次
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
// 如果是精准一次，必须设置 事务的前缀
.setTransactionalIdPrefix("atguigu-")
// 如果是精准一次，必须设置 事务超时时间: 大于checkpoint间隔，小于 max 15分钟
.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10*60*1000+"")
.build();
sensorDS.sinkTo(kafkaSink);
env.execute();
}
}
```
自定义序列化器，实现带key的record:
```java
public class SinkKafkaWithKey {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
env.setRestartStrategy(RestartStrategies.noRestart());
SingleOutputStreamOperator sensorDS = env
.socketTextStream("hadoop102", 7777);
/**
* 如果要指定写入kafka的key，可以自定义序列化器：
* 1、实现 一个接口，重写 序列化 方法
* 2、指定key，转成 字节数组
* 3、指定value，转成 字节数组
* 4、返回一个 ProducerRecord对象，把key、value放进去
*/
KafkaSink kafkaSink = KafkaSink.builder()
.setBootstrapServers("hadoop102:9092,hadoop103:9092,hadoop104:9092")
.setRecordSerializer(
new KafkaRecordSerializationSchema() {
@Nullable
@Override
public ProducerRecord serialize(String element, KafkaSinkContext context, Long timestamp) {
String[] datas = element.split(",");
byte[] key = datas[0].getBytes(StandardCharsets.UTF_8);
byte[] value = element.getBytes(StandardCharsets.UTF_8);
return new ProducerRecord<>("ws", key, value);
}
}
)
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
.setTransactionalIdPrefix("atguigu-")
.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10 * 60 * 1000 + "")
.build();
sensorDS.sinkTo(kafkaSink);
env.execute();
}
}
```
（4）运行代码，在Linux主机启动一个消费者，查看是否收到数据
```shell
[atguigu@hadoop102 ~]$
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic ws
```
#### **5.4.4 输出到MySQL（JDBC）**
写入数据的MySQL的测试步骤如下。
（1）添加依赖
添加MySQL驱动：
```xml
mysql
mysql-connector-java
8.0.27
```
官方还未提供flink-connector-jdbc的1.17.0的正式依赖，暂时从apache snapshot仓库下载，pom文件中指定仓库路径：
```xml
apache-snapshots
apache snapshots
https://repository.apache.org/content/repositories/snapshots/
```
添加依赖：
```xml
org.apache.flink
flink-connector-jdbc
1.17-SNAPSHOT
```
如果不生效，还需要修改本地maven的配置文件，mirrorOf中添加如下标红内容：
```xml
aliyunmaven
*,!apache-snapshots
阿里云公共仓库
https://maven.aliyun.com/repository/public
```
（2）启动MySQL，在test库下建表ws
```mysql
mysql>
CREATE TABLE `ws` (
`id` varchar(100) NOT NULL,
`ts` bigint(20) DEFAULT NULL,
`vc` int(11) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8
```
（3）编写输出到MySQL的示例代码
```java
public class SinkMySQL {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator sensorDS = env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction());
/**
* TODO 写入mysql
* 1、只能用老的sink写法： addsink
* 2、JDBCSink的4个参数:
* 第一个参数： 执行的sql，一般就是 insert into
* 第二个参数： 预编译sql， 对占位符填充值
* 第三个参数： 执行选项 ---》 攒批、重试
* 第四个参数： 连接选项 ---》 url、用户名、密码
*/
SinkFunction jdbcSink = JdbcSink.sink(
"insert into ws values(?,?,?)",
new JdbcStatementBuilder() {
@Override
public void accept(PreparedStatement preparedStatement, WaterSensor waterSensor) throws SQLException {
//每收到一条WaterSensor，如何去填充占位符
preparedStatement.setString(1, waterSensor.getId());
preparedStatement.setLong(2, waterSensor.getTs());
preparedStatement.setInt(3, waterSensor.getVc());
}
},
JdbcExecutionOptions.builder()
.withMaxRetries(3) // 重试次数
.withBatchSize(100) // 批次的大小：条数
.withBatchIntervalMs(3000) // 批次的时间
.build(),
new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl("jdbc:mysql://hadoop102:3306/test?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8")
.withUsername("root")
.withPassword("000000")
.withConnectionCheckTimeoutSeconds(60) // 重试的超时时间
.build()
);
sensorDS.addSink(jdbcSink);
env.execute();
}
}
```
（4）运行代码，用客户端连接MySQL，查看是否成功写入数据。
#### **5.4.5 自定义Sink输出**
如果我们想将数据存储到我们自己的存储设备中，而Flink并没有提供可以直接使用的连接器，就只能自定义Sink进行输出了。与Source类似，Flink为我们提供了通用的SinkFunction接口和对应的RichSinkDunction抽象类，只要实现它，通过简单地调用DataStream的.addSink()方法就可以自定义写入任何外部存储。
```
stream.addSink(new MySinkFunction());
```
在实现SinkFunction的时候，需要重写的一个关键方法invoke()，在这个方法中我们就可以实现将流里的数据发送出去的逻辑。
这种方式比较通用，对于任何外部存储系统都有效；不过自定义Sink想要实现状态一致性并不容易，所以一般只在没有其它选择时使用。实际项目中用到的外部连接器Flink官方基本都已实现，而且在不断地扩充，因此自定义的场景并不常见。
## **第6章 Flink中的时间和窗口**
在批处理统计中，我们可以等待一批数据都到齐后，统一处理。但是在实时处理统计中，我们是来一条就得处理一条，那么我们怎么统计最近一段时间内的数据呢？引入“窗口”。
所谓的“窗口”，一般就是划定的一段时间范围，也就是==“时间窗”==；对在这范围内的数据进行处理，就是所谓的==窗口计算==。所以窗口和时间往往是分不开的。接下来我们就深入了解一下Flink中的时间语义和窗口的应用。
### **6.1 窗口（Window）**
#### **6.1.1 窗口的概念**
Flink是一种流式计算引擎，主要是来处理无界数据流的，数据源源不断、无穷无尽。想要更加方便高效地处理无界流，一种方式就是将无限数据切割成有限的“数据块”进行处理，这就是所谓的“窗口”（Window）。
**注意：**Flink中窗口并不是静态准备好的，而是==动态创建==——当有落在这个窗口区间范围的数据达到时，才创建对应的窗口。另外，这里我们认为到达窗口结束时间时，窗口就触发计算并关闭，事实上“触发计算”和“窗口关闭”两个行为也可以分开，这部分内容我们会在后面详述。
#### **6.1.2 窗口的分类**
我们在上一节举的例子，其实是最为简单的一种时间窗口。在Flink中，窗口的应用非常灵活，我们可以使用各种不同类型的窗口来实现需求。接下来我们就从不同的角度，对Flink中内置的窗口做一个分类说明。
**1）按照驱动类型分**
**2）按照窗口分配数据的规则分类**
根据分配数据的规则，窗口的具体实现可以分为4类：滚动窗口（Tumbling Window）、滑动窗口（Sliding Window）、会话窗口（Session Window），以及全局窗口（Global Window）。
#### **6.1.3 窗口API概览**
**1）按键分区（Keyed）和非按键分区（Non-Keyed）**
在定义窗口操作之前，首先需要确定，到底是基于按键分区（Keyed）的数据流KeyedStream来开窗，还是直接在没有按键分区的DataStream上开窗。也就是说，在调用窗口算子之前，是否有keyBy操作。
（1）按键分区窗口（Keyed Windows）
经过按键分区keyBy操作后，数据流会按照key被分为多条逻辑流（logical streams），这就是KeyedStream。基于KeyedStream进行窗口操作时，窗口计算会在多个并行子任务上同时执行。相同key的数据会被发送到同一个并行子任务，而窗口操作会基于每个key进行单独的处理。所以可以认为，每个key上都定义了一组窗口，各自独立地进行统计计算。
在代码实现上，我们需要先对DataStream调用.keyBy()进行按键分区，然后再调用.window()定义窗口。
```
stream.keyBy(...)
.window(...)
```
（2）非按键分区（Non-Keyed Windows）
如果没有进行keyBy，那么原始的DataStream就不会分成多条逻辑流。这时窗口逻辑只能在一个任务（task）上执行，就相当于并行度变成了1。
在代码中，直接基于DataStream调用.windowAll()定义窗口。
```
stream.windowAll(...)
```
注意：对于==非按键分区的窗口操作，手动调大窗口算子的并行度也是无效的==，windowAll本身就是一个非并行的操作。
**2）代码中窗口API的调用**
窗口操作主要有两个部分：窗口分配器（Window Assigners）和窗口函数（Window Functions）。
```
stream.keyBy()
.window()
.aggregate()
```
其中.window()方法需要传入一个窗口分配器，它指明了窗口的类型；而后面的.aggregate()方法传入一个窗口函数作为参数，它用来定义窗口具体的处理逻辑。窗口分配器有各种形式，而窗口函数的调用方法也不只.aggregate()一种，我们接下来就详细展开讲解。
#### **6.1.4 窗口分配器**
定义窗口分配器（Window Assigners）是构建窗口算子的第一步，它的作用就是定义数据应该被“分配”到哪个窗口。所以可以说，窗口分配器其实就是在指定窗口的类型。
窗口分配器最通用的定义方式，就是调用.window()方法。这个方法需要传入一个WindowAssigner作为参数，返回WindowedStream。如果是非按键分区窗口，那么直接调用.windowAll()方法，同样传入一个WindowAssigner，返回的是AllWindowedStream。
窗口按照驱动类型可以分成时间窗口和计数窗口，而按照具体的分配规则，又有滚动窗口、滑动窗口、会话窗口、全局窗口四种。除去需要自定义的全局窗口外，其他常用的类型Flink中都给出了内置的分配器实现，我们可以方便地调用实现各种需求。
##### **6.1.4.1 时间窗口**
时间窗口是最常用的窗口类型，又可以细分为滚动、滑动和会话三种。
**（1）滚动处理时间窗口**
窗口分配器由类TumblingProcessingTimeWindows提供，需要调用它的静态方法.of()。
```
stream.keyBy(...)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.aggregate(...)
```
这里.of()方法需要传入一个Time类型的参数size，表示滚动窗口的大小，我们这里创建了一个长度为5秒的滚动窗口。
另外，.of()还有一个重载方法，可以传入两个Time类型的参数：size和offset。第一个参数当然还是窗口大小，第二个参数则表示窗口起始点的偏移量。
**（2）滑动处理时间窗口**
窗口分配器由类SlidingProcessingTimeWindows提供，同样需要调用它的静态方法.of()。
```
stream.keyBy(...)
.window(SlidingProcessingTimeWindows.of(Time.seconds(10)，Time.seconds(5)))
.aggregate(...)
```
这里.of()方法需要传入两个Time类型的参数：size和slide，前者表示滑动窗口的大小，后者表示滑动窗口的滑动步长。我们这里创建了一个长度为10秒、滑动步长为5秒的滑动窗口。
滑动窗口同样可以追加第三个参数，用于指定窗口起始点的偏移量，用法与滚动窗口完全一致。
**（3）处理时间会话窗口**
窗口分配器由类ProcessingTimeSessionWindows提供，需要调用它的静态方法.withGap()或者.withDynamicGap()。
```
stream.keyBy(...)
.window(ProcessingTimeSessionWindows.withGap(Time.seconds(10)))
.aggregate(...)
```
这里.withGap()方法需要传入一个Time类型的参数size，表示会话的超时时间，也就是最小间隔session gap。我们这里创建了静态会话超时时间为10秒的会话窗口。
另外，还可以调用withDynamicGap()方法定义session gap的动态提取逻辑。
**（4）滚动事件时间窗口**
窗口分配器由类TumblingEventTimeWindows提供，用法与滚动处理事件窗口完全一致。
```
stream.keyBy(...)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.aggregate(...)
```
**（5）滑动事件时间窗口**
窗口分配器由类SlidingEventTimeWindows提供，用法与滑动处理事件窗口完全一致。
```
stream.keyBy(...)
.window(SlidingEventTimeWindows.of(Time.seconds(10)，Time.seconds(5)))
.aggregate(...)
```
**（6）事件时间会话窗口**
窗口分配器由类EventTimeSessionWindows提供，用法与处理事件会话窗口完全一致。
```
stream.keyBy(...)
.window(EventTimeSessionWindows.withGap(Time.seconds(10)))
.aggregate(...)
```
##### **6.1.4.2 计数窗口**
计数窗口概念非常简单，本身底层是基于全局窗口（Global Window）实现的。Flink为我们提供了非常方便的接口：直接调用.countWindow()方法。根据分配规则的不同，又可以分为滚动计数窗口和滑动计数窗口两类，下面我们就来看它们的具体实现。
**（1）滚动计数窗口**
滚动计数窗口只需要传入一个长整型的参数size，表示窗口的大小。
```
stream.keyBy(...)
.countWindow(10)
```
我们定义了一个长度为10的滚动计数窗口，当窗口中元素数量达到10的时候，就会触发计算执行并关闭窗口。
**（2）滑动计数窗口**
与滚动计数窗口类似，不过需要在.countWindow()调用时传入两个参数：size和slide，前者表示窗口大小，后者表示滑动步长。
```
stream.keyBy(...)
.countWindow(10，3)
```
我们定义了一个长度为10、滑动步长为3的滑动计数窗口。每个窗口统计10个数据，每隔3个数据就统计输出一次结果。
3）全局窗口
全局窗口是计数窗口的底层实现，一般在需要自定义窗口时使用。它的定义同样是直接调用.window()，分配器由GlobalWindows类提供。
```
stream.keyBy(...)
.window(GlobalWindows.create());
```
需要注意使用全局窗口，必须自行定义触发器才能实现窗口计算，否则起不到任何作用。
#### **6.1.5 窗口函数**
窗口函数定义了要对窗口中收集的数据做的计算操作，根据处理的方式可以分为两类：增量聚合函数和全窗口函数。下面我们来进行分别讲解。
##### **6.1.5.1 增量聚合函数（ReduceFunction / AggregateFunction）**
窗口将数据收集起来，最基本的处理操作当然就是进行聚合。我们可以每来一个数据就在之前结果上聚合一次，这就是“增量聚合”。
典型的增量聚合函数有两个：ReduceFunction和AggregateFunction。
**1）归约函数（ReduceFunction）**
代码示例：
```java
public class WindowReduceDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction())
.keyBy(r -> r.getId())
// 设置滚动事件时间窗口
.window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
.reduce(new ReduceFunction() {
@Override
public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
System.out.println("调用reduce方法，之前的结果:"+value1 + ",现在来的数据:"+value2);
return new WaterSensor(value1.getId(), System.currentTimeMillis(),value1.getVc()+value2.getVc());
}
})
.print();
env.execute();
}
}
```
**2）聚合函数（AggregateFunction）**
ReduceFunction可以解决大多数归约聚合的问题，但是这个接口有一个限制，就是聚合状态的类型、输出结果的类型都必须和输入数据类型一样。
Flink Window API中的aggregate就突破了这个限制，可以定义更加灵活的窗口聚合操作。这个方法需要传入一个AggregateFunction的实现类作为参数。
AggregateFunction可以看作是ReduceFunction的通用版本，这里有三种类型：输入类型（IN）、累加器类型（ACC）和输出类型（OUT）。输入类型IN就是输入流中元素的数据类型；累加器类型ACC则是我们进行聚合的中间状态类型；而输出类型当然就是最终计算结果的类型了。
接口中有四个方法：
- createAccumulator()：创建一个累加器，这就是为聚合创建了一个初始状态，每个聚合任务只会调用一次。
- add()：将输入的元素添加到累加器中。
- getResult()：从累加器中提取聚合的输出结果。
- merge()：合并两个累加器，并将合并后的状态作为一个累加器返回。
所以可以看到，AggregateFunction的工作原理是：首先调用createAccumulator()为任务初始化一个状态（累加器）；而后每来一个数据就调用一次add()方法，对数据进行聚合，得到的结果保存在状态中；等到了窗口需要输出时，再调用getResult()方法得到计算结果。很明显，与ReduceFunction相同，AggregateFunction也是增量式的聚合；而由于输入、中间状态、输出的类型可以不同，使得应用更加灵活方便。
代码实现如下：
```java
public class WindowAggregateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator sensorDS = env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction());
KeyedStream sensorKS = sensorDS.keyBy(sensor -> sensor.getId());
// 1. 窗口分配器
WindowedStream sensorWS = sensorKS.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
SingleOutputStreamOperator aggregate = sensorWS
.aggregate(
new AggregateFunction() {
@Override
public Integer createAccumulator() {
System.out.println("创建累加器");
return 0;
}
@Override
public Integer add(WaterSensor value, Integer accumulator) {
System.out.println("调用add方法,value="+value);
return accumulator + value.getVc();
}
@Override
public String getResult(Integer accumulator) {
System.out.println("调用getResult方法");
return accumulator.toString();
}
@Override
public Integer merge(Integer a, Integer b) {
System.out.println("调用merge方法");
return null;
}
}
);
aggregate.print();
env.execute();
}
}
```
另外，Flink也为窗口的聚合提供了一系列预定义的简单聚合方法，可以直接基于WindowedStream调用。主要包括.sum()/max()/maxBy()/min()/minBy()，与KeyedStream的简单聚合非常相似。它们的底层，其实都是通过AggregateFunction来实现的。
##### **6.1.5.2 全窗口函数（full window functions）**
有些场景下，我们要做的计算必须基于全部的数据才有效，这时做增量聚合就没什么意义了；另外，输出的结果有可能要包含上下文中的一些信息（比如窗口的起始时间），这是增量聚合函数做不到的。
所以，我们还需要有更丰富的窗口计算方式。窗口操作中的另一大类就是全窗口函数。与增量聚合函数不同，全窗口函数需要先收集窗口中的数据，并在内部缓存起来，等到窗口要输出结果的时候再取出数据进行计算。
在Flink中，全窗口函数也有两种：WindowFunction和ProcessWindowFunction。
**1）窗口函数（WindowFunction）**
WindowFunction字面上就是“窗口函数”，它其实是老版本的通用窗口函数接口。我们可以基于WindowedStream调用.apply()方法，传入一个WindowFunction的实现类。
```
stream
.keyBy()
.window()
.apply(new MyWindowFunction());
```
这个类中可以获取到包含窗口所有数据的可迭代集合（Iterable），还可以拿到窗口（Window）本身的信息。
不过WindowFunction能提供的上下文信息较少，也没有更高级的功能。==事实上，它的作用可以被ProcessWindowFunction全覆盖，所以之后可能会逐渐弃用。==
**2）处理窗口函数（ProcessWindowFunction）**
ProcessWindowFunction是Window API中最底层的通用窗口函数接口。之所以说它“最底层”，是因为除了可以拿到窗口中的所有数据之外，ProcessWindowFunction还可以获取到一个“上下文对象”（Context）。这个上下文对象非常强大，不仅能够获取窗口信息，还可以访问当前的时间和状态信息。这里的时间就包括了处理时间（processing time）和事件时间水位线（event time watermark）。这就使得ProcessWindowFunction更加灵活、功能更加丰富，其实就是一个增强版的WindowFunction。
事实上，ProcessWindowFunction是Flink底层API——处理函数（process function）中的一员，关于处理函数我们会在后续章节展开讲解。
代码实现如下：
```java
public class WindowProcessDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator sensorDS = env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction());
KeyedStream sensorKS = sensorDS.keyBy(sensor -> sensor.getId());
// 1. 窗口分配器
WindowedStream sensorWS = sensorKS.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
SingleOutputStreamOperator process = sensorWS
.process(
new ProcessWindowFunction() {
@Override
public void process(String s, Context context, Iterable elements, Collector out) throws Exception {
long count = elements.spliterator().estimateSize();
long windowStartTs = context.window().getStart();
long windowEndTs = context.window().getEnd();
String windowStart = DateFormatUtils.format(windowStartTs, "yyyy-MM-dd HH:mm:ss.SSS");
String windowEnd = DateFormatUtils.format(windowEndTs, "yyyy-MM-dd HH:mm:ss.SSS");
out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());
}
}
);
process.print();
env.execute();
}
}
```
##### **6.1.5.3 增量聚合和全窗口函数的结合使用**
在实际应用中，我们往往希望兼具这两者的优点，把它们结合在一起使用。Flink的Window API就给我们实现了这样的用法。
我们之前在调用WindowedStream的.reduce()和.aggregate()方法时，只是简单地直接传入了一个ReduceFunction或AggregateFunction进行增量聚合。除此之外，其实还可以传入第二个参数：一个全窗口函数，可以是WindowFunction或者ProcessWindowFunction。
```java
// ReduceFunction与WindowFunction结合
public SingleOutputStreamOperator reduce(
ReduceFunction reduceFunction，WindowFunction function)
// ReduceFunction与ProcessWindowFunction结合
public SingleOutputStreamOperator reduce(
ReduceFunction reduceFunction，ProcessWindowFunction function)
// AggregateFunction与WindowFunction结合
public SingleOutputStreamOperator aggregate(
AggregateFunction aggFunction，WindowFunction windowFunction)
// AggregateFunction与ProcessWindowFunction结合
public SingleOutputStreamOperator aggregate(
AggregateFunction aggFunction,
ProcessWindowFunction windowFunction)
```
这样调用的处理机制是：基于第一个参数（增量聚合函数）来处理窗口数据，每来一个数据就做一次聚合；等到窗口需要触发计算时，则调用第二个参数（全窗口函数）的处理逻辑输出结果。需要注意的是，这里的全窗口函数就不再缓存所有数据了，而是直接将增量聚合函数的结果拿来当作了Iterable类型的输入。
具体实现代码如下：
```java
public class WindowAggregateAndProcessDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator sensorDS = env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction());
KeyedStream sensorKS = sensorDS.keyBy(sensor -> sensor.getId());
// 1. 窗口分配器
WindowedStream sensorWS = sensorKS.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
// 2. 窗口函数：
/**
* 增量聚合 Aggregate + 全窗口 process
* 1、增量聚合函数处理数据： 来一条计算一条
* 2、窗口触发时， 增量聚合的结果（只有一条） 传递给 全窗口函数
* 3、经过全窗口函数的处理包装后，输出
*
* 结合两者的优点：
* 1、增量聚合： 来一条计算一条，存储中间的计算结果，占用的空间少
* 2、全窗口函数： 可以通过 上下文 实现灵活的功能
*/
// sensorWS.reduce() //也可以传两个
SingleOutputStreamOperator result = sensorWS.aggregate(
new MyAgg(),
new MyProcess()
);
result.print();
env.execute();
}
public static class MyAgg implements AggregateFunction{
@Override
public Integer createAccumulator() {
System.out.println("创建累加器");
return 0;
}
@Override
public Integer add(WaterSensor value, Integer accumulator) {
System.out.println("调用add方法,value="+value);
return accumulator + value.getVc();
}
@Override
public String getResult(Integer accumulator) {
System.out.println("调用getResult方法");
return accumulator.toString();
}
@Override
public Integer merge(Integer a, Integer b) {
System.out.println("调用merge方法");
return null;
}
}
// 全窗口函数的输入类型 = 增量聚合函数的输出类型
public static class MyProcess extends ProcessWindowFunction{
@Override
public void process(String s, Context context, Iterable elements, Collector out) throws Exception {
long startTs = context.window().getStart();
long endTs = context.window().getEnd();
String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
long count = elements.spliterator().estimateSize();
out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());
}
}
}
```
这里我们为了方便处理，单独定义了一个POJO类UrlViewCount来表示聚合输出结果的数据类型，包含了url、浏览量以及窗口的起始结束时间。用一个AggregateFunction来实现增量聚合，每来一个数据就计数加一；得到的结果交给ProcessWindowFunction，结合窗口信息包装成我们想要的UrlViewCount，最终输出统计结果。
#### **6.1.6** **其他API**
对于一个窗口算子而言，窗口分配器和窗口函数是必不可少的。除此之外，Flink还提供了其他一些可选的API，让我们可以更加灵活地控制窗口行为。
##### **6.1.6.1触发器（Trigger）**
触发器主要是用来控制窗口什么时候触发计算。所谓的“触发计算”，本质上就是执行窗口函数，所以可以认为是计算得到结果并输出的过程。
基于WindowedStream调用.trigger()方法，就可以传入一个自定义的窗口触发器（Trigger）。
```
stream.keyBy(...)
.window(...)
.trigger(new MyTrigger())
```
##### **6.1.6.2 移除器（Evictor）**
移除器主要用来定义移除某些数据的逻辑。基于WindowedStream调用.evictor()方法，就可以传入一个自定义的移除器（Evictor）。Evictor是一个接口，不同的窗口类型都有各自预实现的移除器。
```
stream.keyBy(...)
.window(...)
.evictor(new MyEvictor())
```
### **6.2 时间语义**
#### **6.2.1 Flink中的时间语义**
#### **6.2.2 哪种时间语义更重要**
**1）从《星球大战》说起**
为了更加清晰地说明两种语义的区别，我们来举一个非常经典的例子：电影《星球大战》。
如上图所示，我们会发现，看电影其实就是处理影片中数据的过程，所以影片的上映时间就相当于==“处理时间”==；而影片的数据就是所描述的故事，它所发生的背景时间就相当于==“事件时间”==。两种时间语义都有各自的用途，适用于不同的场景。
**2）数据处理系统中的时间语义**
==在实际应用中，事件时间语义会更为常见。==一般情况下，==业务日志数据中都会记录数据生成的时间戳（timestamp），它就可以作为事件时间的判断基础。==
在Flink中，由于处理时间比较简单，早期版本默认的时间语义是处理时间；而考虑到事件时间在实际应用中更为广泛，从Flink1.12版本开始，==Flink已经将事件时间作为默认的时间语义了。==
### **6.3 水位线（Watermark）**
#### **6.3.1 事件时间和窗口**
#### **6.3.2 什么是水位线**
在Flink中，用来衡量==事件时间==进展的标记，就被称作==“水位线”（Watermark）==。
具体实现上，水位线可以看作一条特殊的数据记录，它是插入到数据流中的一个标记点，主要内容就是一个时间戳，用来指示当前的事件时间。而它插入流中的位置，就应该是在某个数据到来之后；这样就可以从这个数据中提取时间戳，作为当前水位线的时间戳了。
#### **6.3.3 水位线和窗口的工作原理**
**注意：**Flink中窗口并不是静态准备好的，而是动态创建——当有落在这个窗口区间范围的数据达到时，才创建对应的窗口。另外，这里我们认为到达窗口结束时间时，窗口就触发计算并关闭，事实上“触发计算”和“窗口关闭”两个行为也可以分开，这部分内容我们会在后面详述。
#### **6.3.4 生成水位线**
##### **6.3.4.1 生成水位线的总体原则**
完美的水位线是“绝对正确”的，也就是一个水位线一旦出现，就表示这个时间之前的数据已经全部到齐、之后再也不会出现了。不过==如果要保证绝对正确，就必须等足够长的时间，这会带来更高的延迟。==
==如果我们希望处理得更快、实时性更强，那么可以将水位线延迟设得低一些。==这种情况下，可能很多迟到数据会在水位线之后才到达，就会导致窗口遗漏数据，计算结果不准确。当然，==如果我们对准确性完全不考虑、一味地追求处理速度，可以直接使用处理时间语义，这在理论上可以得到最低的延迟。==
所以Flink中的水位线，其实是流处理中==对低延迟和结果正确性的一个权衡机制==，而且把控制的权力交给了程序员，我们可以在代码中定义水位线的生成策略。
##### **6.3.4.2 水位线生成策略**
在Flink的DataStream API中，有一个单独用于生成水位线的方法：.assignTimestampsAndWatermarks()，它主要用来为流中的数据分配时间戳，并生成水位线来指示事件时间。具体使用如下：
```java
DataStream stream = env.addSource(new ClickSource());
DataStream withTimestampsAndWatermarks =
stream.assignTimestampsAndWatermarks();
```
说明：WatermarkStrategy作为参数，这就是所谓的“水位线生成策略”。WatermarkStrategy是一个接口，该接口中包含了一个“时间戳分配器”TimestampAssigner和一个“水位线生成器”WatermarkGenerator。
```java
public interface WatermarkStrategy
extends TimestampAssignerSupplier,
WatermarkGeneratorSupplier{
// 负责从流中数据元素的某个字段中提取时间戳，并分配给元素。时间戳的分配是生成水位线的基础。
@Override
TimestampAssigner createTimestampAssigner(TimestampAssignerSupplier.Context context);
// 主要负责按照既定的方式，基于时间戳生成水位线
@Override
WatermarkGenerator createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);
}
```
##### **6.3.4.3 Flink内置水位线**
**1）有序流中内置水位线设置**
对于有序流，主要特点就是时间戳单调增长，所以永远不会出现迟到数据的问题。这是周期性生成水位线的最简单的场景，直接调用WatermarkStrategy.forMonotonousTimestamps()方法就可以实现。
```java
public class WatermarkMonoDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator sensorDS = env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction());
// TODO 1.定义Watermark策略
WatermarkStrategy watermarkStrategy = WatermarkStrategy
// 1.1 指定watermark生成：升序的watermark，没有等待时间
.forMonotonousTimestamps()
// 1.2 指定 时间戳分配器，从数据中提取
.withTimestampAssigner(new SerializableTimestampAssigner() {
@Override
public long extractTimestamp(WaterSensor element, long recordTimestamp) {
// 返回的时间戳，要 毫秒
System.out.println("数据=" + element + ",recordTs=" + recordTimestamp);
return element.getTs() * 1000L;
}
});
// TODO 2. 指定 watermark策略
SingleOutputStreamOperator sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);
sensorDSwithWatermark.keyBy(sensor -> sensor.getId())
// TODO 3.使用 事件时间语义 的窗口
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.process(
new ProcessWindowFunction() {
@Override
public void process(String s, Context context, Iterable elements, Collector out) throws Exception {
long startTs = context.window().getStart();
long endTs = context.window().getEnd();
String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
long count = elements.spliterator().estimateSize();
out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());
}
}
)
.print();
env.execute();
}
}
```
**2）乱序流中内置水位线设置**
由于乱序流中需要等待迟到数据到齐，所以必须设置一个固定量的延迟时间。这时生成水位线的时间戳，就是当前数据流中最大的时间戳减去延迟的结果，相当于把表调慢，当前时钟会滞后于数据的最大时间戳。调用WatermarkStrategy. ==forBoundedOutOfOrderness==()方法就可以实现。这个方法需要传入一个maxOutOfOrderness参数，表示“最大乱序程度”，它表示数据流中乱序数据时间戳的最大差值；如果我们能确定乱序程度，那么设置对应时间长度的延迟，就可以等到所有的乱序数据了。
```java
public class WatermarkOutOfOrdernessDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator sensorDS = env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction());
// TODO 1.定义Watermark策略
WatermarkStrategy watermarkStrategy = WatermarkStrategy
// 1.1 指定watermark生成：乱序的，等待3s
.forBoundedOutOfOrderness(Duration.ofSeconds(3))
// 1.2 指定 时间戳分配器，从数据中提取
.withTimestampAssigner(
(element, recordTimestamp) -> {
// 返回的时间戳，要 毫秒
System.out.println("数据=" + element + ",recordTs=" + recordTimestamp);
return element.getTs() * 1000L;
});
// TODO 2. 指定 watermark策略
SingleOutputStreamOperator sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);
sensorDSwithWatermark.keyBy(sensor -> sensor.getId())
// TODO 3.使用 事件时间语义 的窗口
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.process(
new ProcessWindowFunction() {
@Override
public void process(String s, Context context, Iterable elements, Collector out) throws Exception {
long startTs = context.window().getStart();
long endTs = context.window().getEnd();
String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
long count = elements.spliterator().estimateSize();
out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());
}
}
)
.print();
env.execute();
}
}
```
##### **6.3.4.4 自定义水位线生成器**
**1）周期性水位线生成器（Periodic Generator）**
周期性生成器一般是通过onEvent()观察判断输入的事件，而在onPeriodicEmit()里发出水位线。
下面是一段自定义周期性生成水位线的代码：
```java
import com.atguigu.bean.Event;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
// 自定义水位线的产生
public class CustomPeriodicWatermarkExample {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env
.addSource(new ClickSource())
.assignTimestampsAndWatermarks(new CustomWatermarkStrategy())
.print();
env.execute();
}
public static class CustomWatermarkStrategy implements WatermarkStrategy {
@Override
public TimestampAssigner createTimestampAssigner(TimestampAssignerSupplier.Context context) {
return new SerializableTimestampAssigner() {
@Override
public long extractTimestamp(Event element，long recordTimestamp) {
return element.timestamp; // 告诉程序数据源里的时间戳是哪一个字段
}
};
}
@Override
public WatermarkGenerator createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
return new CustomBoundedOutOfOrdernessGenerator();
}
}
public static class CustomBoundedOutOfOrdernessGenerator implements WatermarkGenerator {
private Long delayTime = 5000L; // 延迟时间
private Long maxTs = -Long.MAX_VALUE + delayTime + 1L; // 观察到的最大时间戳
@Override
public void onEvent(Event event，long eventTimestamp，WatermarkOutput output) {
// 每来一条数据就调用一次
maxTs = Math.max(event.timestamp，maxTs); // 更新最大时间戳
}
@Override
public void onPeriodicEmit(WatermarkOutput output) {
// 发射水位线，默认200ms调用一次
output.emitWatermark(new Watermark(maxTs - delayTime - 1L));
}
}
}
```
我们在onPeriodicEmit()里调用output.emitWatermark()，就可以发出水位线了；这个方法由系统框架周期性地调用，默认200ms一次。
如果想修改默认周期时间，可以通过下面方法修改。例如：修改为400ms
```
env.getConfig().setAutoWatermarkInterval(400L);
```
**2）断点式水位线生成器（Punctuated Generator）**
断点式生成器会不停地检测onEvent()中的事件，当发现带有水位线信息的事件时，就立即发出水位线。我们把发射水位线的逻辑写在onEvent方法当中即可。
**3）在数据源中发送水位线**
我们也可以在自定义的数据源中抽取事件时间，然后发送水位线。这里要注意的是，在自定义数据源中发送了水位线以后，就不能再在程序中使用assignTimestampsAndWatermarks方法来生成水位线了。在自定义数据源中生成水位线和在程序中使用assignTimestampsAndWatermarks方法生成水位线二者只能取其一。示例程序如下：
```
env.fromSource(
kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)), "kafkasource"
)
```
#### **6.3.5 水位线的传递**
在流处理中，上游任务处理完水位线、时钟改变之后，要把当前的水位线再次发出，广播给所有的下游子任务。而当一个任务接收到多个上游并行任务传递来的水位线时，应该==以最小的那个作为当前任务的事件时钟。==
水位线在上下游任务之间的传递，非常巧妙地避免了分布式系统中没有统一时钟的问题，==每个任务都以“处理完之前所有数据”为标准来确定自己的时钟。==
案例：6.3.4.3 中乱序流的watermark，将并行度设为2，观察现象。
在多个上游并行任务中，如果有其中一个没有数据，由于当前Task是以最小的那个作为当前任务的事件时钟，就会导致当前Task的水位线无法推进，就可能导致窗口无法触发。这时候可以设置空闲等待。
用5.3.4.6中的自定义分区器，只输入奇数来模拟部分subtask无数据，代码如下：
```java
public class WatermarkIdlenessDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
// 自定义分区器：数据%分区数，只输入奇数，都只会去往map的一个子任务
SingleOutputStreamOperator socketDS = env
.socketTextStream("hadoop102", 7777)
.partitionCustom(new MyPartitioner(), r -> r)
.map(r -> Integer.parseInt(r))
.assignTimestampsAndWatermarks(
WatermarkStrategy
.forMonotonousTimestamps()
.withTimestampAssigner((r, ts) -> r * 1000L)
.withIdleness(Duration.ofSeconds(5)) //空闲等待5s
);
// 分成两组： 奇数一组，偶数一组 ， 开10s的事件时间滚动窗口
socketDS
.keyBy(r -> r % 2)
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.process(new ProcessWindowFunction() {
@Override
public void process(Integer integer, Context context, Iterable elements, Collector out) throws Exception {
long startTs = context.window().getStart();
long endTs = context.window().getEnd();
String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
long count = elements.spliterator().estimateSize();
out.collect("key=" + integer + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());
}
})
.print();
env.execute();
}
}
```
#### **6.3.6** **迟到数据的处理**
##### **6.3.6.1 推迟水印推进**
在水印产生时，设置一个乱序容忍度，推迟系统时间的推进，保证窗口计算被延迟执行，为乱序的数据争取更多的时间进入窗口。
```
WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10));
```
##### **6.3.6.2 设置窗口延迟关闭**
Flink的窗口，也允许迟到数据。当触发了窗口计算后，会先计算当前的结果，但是此时并不会关闭窗口。
以后每来一条迟到数据，就触发一次这条数据所在窗口计算(增量计算)。直到wartermark 超过了窗口结束时间+推迟时间，此时窗口会真正关闭。
```
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.allowedLateness(Time.seconds(3))
```
注意:
允许迟到只能运用在event time上
##### **6.3.6.3 使用侧流接收迟到的数据**
```
.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
.allowedLateness(Time.seconds(3))
.sideOutputLateData(lateWS)
```
完整案例代码如下：
```java
public class WatermarkLateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator sensorDS = env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction());
WatermarkStrategy watermarkStrategy = WatermarkStrategy
.forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner((element, recordTimestamp) -> element.getTs() * 1000L);
SingleOutputStreamOperator sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);
OutputTag lateTag = new OutputTag<>("late-data", Types.POJO(WaterSensor.class));
SingleOutputStreamOperator process = sensorDSwithWatermark.keyBy(sensor -> sensor.getId())
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.allowedLateness(Time.seconds(2)) // 推迟2s关窗
.sideOutputLateData(lateTag) // 关窗后的迟到数据，放入侧输出流
.process(
new ProcessWindowFunction() {
@Override
public void process(String s, Context context, Iterable elements, Collector out) throws Exception {
long startTs = context.window().getStart();
long endTs = context.window().getEnd();
String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
long count = elements.spliterator().estimateSize();
out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());
}
}
);
process.print();
// 从主流获取侧输出流，打印
process.getSideOutput(lateTag).printToErr("关窗后的迟到数据");
env.execute();
}
}
```
### **6.4** **基于时间的合流——双流联结（Join）**
可以发现，根据某个key合并两条流，与关系型数据库中表的join操作非常相近。事实上，Flink中两条流的connect操作，就可以通过keyBy指定键进行分组后合并，实现了类似于SQL中的join操作；另外connect支持处理函数，可以使用自定义实现各种需求，其实已经能够处理双流join的大多数场景。
不过处理函数是底层接口，所以尽管connect能做的事情多，但在一些具体应用场景下还是显得太过抽象了。比如，如果我们希望统计固定时间内两条流数据的匹配情况，那就需要自定义来实现——其实这完全可以用窗口（window）来表示。为了更方便地实现基于时间的合流操作，Flink的DataStrema API提供了内置的join算子。
#### **6.4.1 窗口联结（Window Join）**
Flink为基于一段时间的双流合并专门提供了一个窗口联结算子，可以定义时间窗口，并将两条流中共享一个公共键（key）的数据放在窗口中进行配对处理。
**1）窗口联结的调用**
窗口联结在代码中的实现，首先需要调用DataStream的.join()方法来合并两条流，得到一个JoinedStreams；接着通过.where()和.equalTo()方法指定两条流中联结的key；然后通过.window()开窗口，并调用.apply()传入联结窗口函数进行处理计算。通用调用形式如下：
```
stream1.join(stream2)
.where()
.equalTo()
.window()
.apply()
```
上面代码中.where()的参数是键选择器（KeySelector），用来指定第一条流中的key；而.equalTo()传入的KeySelector则指定了第二条流中的key。两者相同的元素，如果在同一窗口中，就可以匹配起来，并通过一个“联结函数”（JoinFunction）进行处理了。
这里.window()传入的就是窗口分配器，之前讲到的三种时间窗口都可以用在这里：滚动窗口（tumbling window）、滑动窗口（sliding window）和会话窗口（session window）。
而后面调用.apply()可以看作实现了一个特殊的窗口函数。注意这里只能调用.apply()，没有其他替代的方法。
传入的JoinFunction也是一个函数类接口，使用时需要实现内部的.join()方法。这个方法有两个参数，分别表示两条流中成对匹配的数据。
其实仔细观察可以发现，窗口join的调用语法和我们熟悉的SQL中表的join非常相似：
```sql
SELECT * FROM table1 t1, table2 t2 WHERE t1.id = t2.id;
```
这句SQL中where子句的表达，等价于inner join ... on，所以本身表示的是两张表基于id的“内连接”（inner join）。而Flink中的window join，同样类似于inner join。也就是说，最后处理输出的，只有两条流中数据按key配对成功的那些；如果某个窗口中一条流的数据没有任何另一条流的数据匹配，那么就不会调用JoinFunction的.join()方法，也就没有任何输出了。
**2）窗口联结实例**
代码实现：
```java
public class WindowJoinDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator> ds1 = env
.fromElements(
Tuple2.of("a", 1),
Tuple2.of("a", 2),
Tuple2.of("b", 3),
Tuple2.of("c", 4)
)
.assignTimestampsAndWatermarks(
WatermarkStrategy
.>forMonotonousTimestamps()
.withTimestampAssigner((value, ts) -> value.f1 * 1000L)
);
SingleOutputStreamOperator> ds2 = env
.fromElements(
Tuple3.of("a", 1,1),
Tuple3.of("a", 11,1),
Tuple3.of("b", 2,1),
Tuple3.of("b", 12,1),
Tuple3.of("c", 14,1),
Tuple3.of("d", 15,1)
)
.assignTimestampsAndWatermarks(
WatermarkStrategy
.>forMonotonousTimestamps()
.withTimestampAssigner((value, ts) -> value.f1 * 1000L)
);
// TODO window join
// 1. 落在同一个时间窗口范围内才能匹配
// 2. 根据keyby的key，来进行匹配关联
// 3. 只能拿到匹配上的数据，类似有固定时间范围的inner join
DataStream join = ds1.join(ds2)
.where(r1 -> r1.f0) // ds1的keyby
.equalTo(r2 -> r2.f0) // ds2的keyby
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.apply(new JoinFunction, Tuple3, String>() {
/**
* 关联上的数据，调用join方法
* @param first ds1的数据
* @param second ds2的数据
* @return
* @throws Exception
*/
@Override
public String join(Tuple2 first, Tuple3 second) throws Exception {
return first + "<----->" + second;
}
});
join.print();
env.execute();
}
}
```
#### **6.4.2 间隔联结（Interval Join）**
在有些场景下，我们要处理的时间间隔可能并不是固定的。这时显然不应该用滚动窗口或滑动窗口来处理——因为匹配的两个数据有可能刚好“卡在”窗口边缘两侧，于是窗口内就都没有匹配了；会话窗口虽然时间不固定，但也明显不适合这个场景。基于时间的窗口联结已经无能为力了。
为了应对这样的需求，Flink提供了一种叫作“间隔联结”（interval join）的合流操作。顾名思义，间隔联结的思路就是针对一条流的每个数据，开辟出其时间戳前后的一段时间间隔，看这期间是否有来自另一条流的数据匹配。
**1）间隔联结的原理**
间隔联结具体的定义方式是，我们给定两个时间点，分别叫作间隔的“上界”（upperBound）和“下界”（lowerBound）；于是对于一条流（不妨叫作A）中的任意一个数据元素a，就可以开辟一段时间间隔：[a.timestamp + lowerBound, a.timestamp + upperBound],即以a的时间戳为中心，下至下界点、上至上界点的一个闭区间：我们就把这段时间作为可以匹配另一条流数据的“窗口”范围。所以对于另一条流（不妨叫B）中的数据元素b，如果它的时间戳落在了这个区间范围内，a和b就可以成功配对，进而进行计算输出结果。所以匹配的条件为：
a.timestamp + lowerBound <= b.timestamp <= a.timestamp + upperBound
这里需要注意，做间隔联结的两条流A和B，也必须基于相同的key；下界lowerBound应该小于等于上界upperBound，两者都可正可负；间隔联结目前只支持事件时间语义。
如下图所示，我们可以清楚地看到间隔联结的方式：
下方的流A去间隔联结上方的流B，所以基于A的每个数据元素，都可以开辟一个间隔区间。我们这里设置下界为-2毫秒，上界为1毫秒。于是对于时间戳为2的A中元素，它的可匹配区间就是[0, 3],流B中有时间戳为0、1的两个元素落在这个范围内，所以就可以得到匹配数据对（2, 0）和（2, 1）。同样地，A中时间戳为3的元素，可匹配区间为[1, 4]，B中只有时间戳为1的一个数据可以匹配，于是得到匹配数据对（3, 1）。
所以我们可以看到，间隔联结同样是一种内连接（inner join）。与窗口联结不同的是，interval join做匹配的时间段是基于流中数据的，所以并不确定；而且流B中的数据可以不只在一个区间内被匹配。
**2）间隔联结的调用**
间隔联结在代码中，是基于KeyedStream的联结（join）操作。DataStream在keyBy得到KeyedStream之后，可以调用.intervalJoin()来合并两条流，传入的参数同样是一个KeyedStream，两者的key类型应该一致；得到的是一个IntervalJoin类型。后续的操作同样是完全固定的：先通过.between()方法指定间隔的上下界，再调用.process()方法，定义对匹配数据对的处理操作。调用.process()需要传入一个处理函数，这是处理函数家族的最后一员：“处理联结函数”ProcessJoinFunction。
通用调用形式如下：
```java
stream1
.keyBy()
.intervalJoin(stream2.keyBy())
.between(Time.milliseconds(-2), Time.milliseconds(1))
.process (new ProcessJoinFunction out) {
out.collect(left + "," + right);
}
});
```
可以看到，抽象类ProcessJoinFunction就像是ProcessFunction和JoinFunction的结合，内部同样有一个抽象方法.processElement()。与其他处理函数不同的是，它多了一个参数，这自然是因为有来自两条流的数据。参数中left指的就是第一条流中的数据，right则是第二条流中与它匹配的数据。每当检测到一组匹配，就会调用这里的.processElement()方法，经处理转换之后输出结果。
**3）间隔联结实例**
**案例需求：**在电商网站中，某些用户行为往往会有短时间内的强关联。我们这里举一个例子，我们有两条流，一条是下订单的流，一条是浏览数据的流。我们可以针对同一个用户，来做这样一个联结。也就是使用一个用户的下订单的事件和这个用户的最近十分钟的浏览数据进行一个联结查询。
（1）代码实现：正常使用
```java
public class IntervalJoinDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator> ds1 = env
.fromElements(
Tuple2.of("a", 1),
Tuple2.of("a", 2),
Tuple2.of("b", 3),
Tuple2.of("c", 4)
)
.assignTimestampsAndWatermarks(
WatermarkStrategy
.>forMonotonousTimestamps()
.withTimestampAssigner((value, ts) -> value.f1 * 1000L)
);
SingleOutputStreamOperator> ds2 = env
.fromElements(
Tuple3.of("a", 1, 1),
Tuple3.of("a", 11, 1),
Tuple3.of("b", 2, 1),
Tuple3.of("b", 12, 1),
Tuple3.of("c", 14, 1),
Tuple3.of("d", 15, 1)
)
.assignTimestampsAndWatermarks(
WatermarkStrategy
.>forMonotonousTimestamps()
.withTimestampAssigner((value, ts) -> value.f1 * 1000L)
);
// TODO interval join
//1. 分别做keyby，key其实就是关联条件
KeyedStream, String> ks1 = ds1.keyBy(r1 -> r1.f0);
KeyedStream, String> ks2 = ds2.keyBy(r2 -> r2.f0);
//2. 调用 interval join
ks1.intervalJoin(ks2)
.between(Time.seconds(-2), Time.seconds(2))
.process(
new ProcessJoinFunction, Tuple3, String>() {
/**
* 两条流的数据匹配上，才会调用这个方法
* @param left ks1的数据
* @param right ks2的数据
* @param ctx 上下文
* @param out 采集器
* @throws Exception
*/
@Override
public void processElement(Tuple2 left, Tuple3 right, Context ctx, Collector out) throws Exception {
// 进入这个方法，是关联上的数据
out.collect(left + "<------>" + right);
}
})
.print();
env.execute();
}
}
```
（2）代码实现：处理迟到数据
```java
public class IntervalJoinWithLateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator> ds1 = env
.socketTextStream("hadoop102", 7777)
.map(new MapFunction>() {
@Override
public Tuple2 map(String value) throws Exception {
String[] datas = value.split(",");
return Tuple2.of(datas[0], Integer.valueOf(datas[1]));
}
})
.assignTimestampsAndWatermarks(
WatermarkStrategy
.>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner((value, ts) -> value.f1 * 1000L)
);
SingleOutputStreamOperator> ds2 = env
.socketTextStream("hadoop102", 8888)
.map(new MapFunction>() {
@Override
public Tuple3 map(String value) throws Exception {
String[] datas = value.split(",");
return Tuple3.of(datas[0], Integer.valueOf(datas[1]), Integer.valueOf(datas[2]));
}
})
.assignTimestampsAndWatermarks(
WatermarkStrategy
.>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner((value, ts) -> value.f1 * 1000L)
);
/**
* TODO Interval join
* 1、只支持事件时间
* 2、指定上界、下界的偏移，负号代表时间往前，正号代表时间往后
* 3、process中，只能处理 join上的数据
* 4、两条流关联后的watermark，以两条流中最小的为准
* 5、如果 当前数据的事件时间 < 当前的watermark，就是迟到数据， 主流的process不处理
* => between后，可以指定将 左流 或 右流 的迟到数据 放入侧输出流
*/
//1. 分别做keyby，key其实就是关联条件
KeyedStream, String> ks1 = ds1.keyBy(r1 -> r1.f0);
KeyedStream, String> ks2 = ds2.keyBy(r2 -> r2.f0);
//2. 调用 interval join
OutputTag> ks1LateTag = new OutputTag<>("ks1-late", Types.TUPLE(Types.STRING, Types.INT));
OutputTag> ks2LateTag = new OutputTag<>("ks2-late", Types.TUPLE(Types.STRING, Types.INT, Types.INT));
SingleOutputStreamOperator process = ks1.intervalJoin(ks2)
.between(Time.seconds(-2), Time.seconds(2))
.sideOutputLeftLateData(ks1LateTag) // 将 ks1的迟到数据，放入侧输出流
.sideOutputRightLateData(ks2LateTag) // 将 ks2的迟到数据，放入侧输出流
.process(
new ProcessJoinFunction, Tuple3, String>() {
/**
* 两条流的数据匹配上，才会调用这个方法
* @param left ks1的数据
* @param right ks2的数据
* @param ctx 上下文
* @param out 采集器
* @throws Exception
*/
@Override
public void processElement(Tuple2 left, Tuple3 right, Context ctx, Collector out) throws Exception {
// 进入这个方法，是关联上的数据
out.collect(left + "<------>" + right);
}
});
process.print("主流");
process.getSideOutput(ks1LateTag).printToErr("ks1迟到数据");
process.getSideOutput(ks2LateTag).printToErr("ks2迟到数据");
env.execute();
}
}
```
## **第7章 处理函数**
之前所介绍的流处理API，无论是基本的转换、聚合，还是更为复杂的窗口操作，其实都是基于DataStream进行转换的，所以可以统称为DataStream API。
在Flink更底层，我们可以不定义任何具体的算子（比如map，filter，或者window），而只是提炼出一个==统一的“处理”（process）操作==——它是所有转换算子的一个概括性的表达，可以==自定义处理逻辑==，所以这一层接口就被叫作==“处理函数”（process function）==。
### **7.1 基本处理函数（ProcessFunction）**
#### **7.1.1** **处理函数的功能和使用**
我们之前学习的转换算子，一般只是针对某种具体操作来定义的，能够拿到的信息比较有限。如果我们想要访问事件的时间戳，或者当前的水位线信息，都是完全做不到的。跟时间相关的操作，目前我们只会用窗口来处理。而在很多应用需求中，要求我们对时间有更精细的控制，需要能够获取水位线，甚至要“把控时间”、定义什么时候做什么事，这就不是基本的时间窗口能够实现的了。
这时就需要使用底层的处理函数。处理函数提供了一个“定时服务”（TimerService），我们可以通过它访问流中的事件（event）、时间戳（timestamp）、水位线（watermark），甚至可以注册“定时事件”。而且处理函数继承了AbstractRichFunction抽象类，所以拥有富函数类的所有特性，同样可以访问状态（state）和其他运行时信息。此外，处理函数还可以直接将数据输出到侧输出流（side output）中。所以，处理函数是最为灵活的处理方法，可以实现各种自定义的业务逻辑。
处理函数的使用与基本的转换操作类似，只需要直接基于DataStream调用.process()方法就可以了。方法需要传入一个ProcessFunction作为参数，用来定义处理逻辑。
```
stream.process(new MyProcessFunction())
```
这里ProcessFunction不是接口，而是一个抽象类，继承了AbstractRichFunction；MyProcessFunction是它的一个具体实现。所以所有的处理函数，都是富函数（RichFunction），富函数可以调用的东西这里同样都可以调用。
#### **7.1.2 ProcessFunction解析**
在源码中我们可以看到，抽象类ProcessFunction继承了AbstractRichFunction，有两个泛型类型参数：I表示Input，也就是输入的数据类型；O表示Output，也就是处理完成之后输出的数据类型。
内部单独定义了两个方法：一个是必须要实现的抽象方法.processElement()；另一个是非抽象方法.onTimer()。
```java
public abstract class ProcessFunction extends AbstractRichFunction {
...
public abstract void processElement(I value, Context ctx, Collector out) throws Exception;
public void onTimer(long timestamp, OnTimerContext ctx, Collector out) throws Exception {}
...
}
```
**1）抽象方法.processElement()**
用于“处理元素”，定义了处理的核心逻辑。这个方法对于流中的==每个元素都会调用一次==，参数包括三个：输入数据值value，上下文ctx，以及“收集器”（Collector）out。方法没有返回值，处理之后的输出数据是通过收集器out来定义的。
- ==value==：当前==流中的输入元素==，也就是正在处理的数据，类型与流中数据类型一致。
- ==ctx==：类型是ProcessFunction中定义的内部抽象类Context，表示当前运行的上下文，可以获取到当前的时间戳，并提供了用于查询时间和注册定时器的“定时服务”（TimerService），以及可以将数据发送到“侧输出流”（side output）的方法.output()。
- ==out==：“收集器”（类型为Collector），用于返回输出数据。使用方式与flatMap算子中的收集器完全一样，直接调用out.collect()方法就可以向下游发出一个数据。这个方法可以多次调用，也可以不调用。
通过几个参数的分析不难发现，ProcessFunction可以轻松实现flatMap、map、filter这样的基本转换功能；而通过富函数提供的获取上下文方法.getRuntimeContext()，也可以自定义状态（state）进行处理，这也就能实现聚合操作的功能了。
**2）非抽象方法.onTimer()**
这个方法只有在注册好的定时器触发的时候才会调用，而定时器是通过“定时服务”TimerService来注册的。打个比方，注册定时器（timer）就是设了一个闹钟，到了设定时间就会响；而.onTimer()中定义的，就是闹钟响的时候要做的事。所以它本质上是一个基于时间的“回调”（callback）方法，通过时间的进展来触发；在事件时间语义下就是由水位线（watermark）来触发了。
定时方法.onTimer()也有三个参数：==时间戳（timestamp），上下文（ctx），以及收集器（out）==。这里的timestamp是指设定好的触发时间，事件时间语义下当然就是水位线了。另外这里同样有上下文和收集器，所以也可以调用定时服务（TimerService），以及任意输出处理之后的数据。
既然有.onTimer()方法做定时触发，我们用ProcessFunction也可以自定义数据按照时间分组、定时触发计算输出结果；这其实就实现了窗口（window）的功能。所以说ProcessFunction其实可以实现一切功能。
==注意：在Flink中，只有“按键分区流”KeyedStream才支持设置定时器的操作。==
#### **7.1.3** **处理函数的分类**
我们知道，DataStream在调用一些转换方法之后，有可能生成新的流类型；例如调用.keyBy()之后得到KeyedStream，进而再调用.window()之后得到WindowedStream。对于不同类型的流，其实==都可以直接调用.process()方法进行自定义处理==，这时传入的参数就都叫作处理函数。当然，它们尽管本质相同，都是可以访问状态和时间信息的底层API，可彼此之间也会有所差异。
Flink提供了8个不同的处理函数：
（1）ProcessFunction
最基本的处理函数，基于DataStream直接调用.process()时作为参数传入。
（2）KeyedProcessFunction
对流按键分区后的处理函数，基于KeyedStream调用.process()时作为参数传入。要想使用定时器，比如基于KeyedStream。
（3）ProcessWindowFunction
开窗之后的处理函数，也是全窗口函数的代表。基于WindowedStream调用.process()时作为参数传入。
（4）ProcessAllWindowFunction
同样是开窗之后的处理函数，基于AllWindowedStream调用.process()时作为参数传入。
（5）CoProcessFunction
合并（connect）两条流之后的处理函数，基于ConnectedStreams调用.process()时作为参数传入。关于流的连接合并操作，我们会在后续章节详细介绍。
（6）ProcessJoinFunction
间隔连接（interval join）两条流之后的处理函数，基于IntervalJoined调用.process()时作为参数传入。
（7）BroadcastProcessFunction
广播连接流处理函数，基于BroadcastConnectedStream调用.process()时作为参数传入。这里的“广播连接流”BroadcastConnectedStream，是一个未keyBy的普通DataStream与一个广播流（BroadcastStream）做连接（conncet）之后的产物。关于广播流的相关操作，我们会在后续章节详细介绍。
（8）KeyedBroadcastProcessFunction
按键分区的广播连接流处理函数，同样是基于BroadcastConnectedStream调用.process()时作为参数传入。与BroadcastProcessFunction不同的是，这时的广播连接流，是一个KeyedStream与广播流（BroadcastStream）做连接之后的产物。
### **7.2 按键分区处理函数（KeyedProcessFunction）**
在上节中提到，==只有在KeyedStream中才支持使用TimerService设置定时器的操作==。所以一般情况下，我们都是先做了keyBy分区之后，再去定义处理操作；代码中更加常见的处理函数是KeyedProcessFunction。
#### **7.2.1** **定时器（Timer）和定时服务（TimerService）**
在.onTimer()方法中可以实现定时处理的逻辑，而它能触发的前提，就是之前曾经注册过定时器、并且现在已经到了触发时间。注册定时器的功能，是通过上下文中提供的“定时服务”来实现的。
定时服务与当前运行的环境有关。前面已经介绍过，ProcessFunction的上下文（Context）中提供了.timerService()方法，可以直接返回一个TimerService对象。==TimerService==是Flink关于时间和定时器的基础服务接口，包含以下六个方法：
```java
// 获取当前的处理时间
long currentProcessingTime();
// 获取当前的水位线（事件时间）
long currentWatermark();
// 注册处理时间定时器，当处理时间超过time时触发
void registerProcessingTimeTimer(long time);
// 注册事件时间定时器，当水位线超过time时触发
void registerEventTimeTimer(long time);
// 删除触发时间为time的处理时间定时器
void deleteProcessingTimeTimer(long time);
// 删除触发时间为time的处理时间定时器
void deleteEventTimeTimer(long time);
```
六个方法可以分成两大类：==基于处理时间和基于事件时间==。而对应的操作主要有三个：获取当前时间，注册定时器，以及删除定时器。需要注意，尽管处理函数中都可以直接访问TimerService，不过只有基于KeyedStream的处理函数，才能去调用注册和删除定时器的方法；未作按键分区的DataStream不支持定时器操作，只能获取当前时间。
TimerService会以键（key）和时间戳为标准，对定时器进行去重；也就是说==对于每个key和时间戳，最多只有一个定时器，如果注册了多次，onTimer()方法也将只被调用一次==。
#### **7.2.2 KeyedProcessFunction案例**
基于keyBy之后的KeyedStream，直接调用.process()方法，这时需要传入的参数就是KeyedProcessFunction的实现类。
```
stream.keyBy( t -> t.f0 )
.process(new MyKeyedProcessFunction())
```
类似地，KeyedProcessFunction也是继承自AbstractRichFunction的一个抽象类，与ProcessFunction的定义几乎完全一样，区别只是在于类型参数多了一个K，这是当前按键分区的key的类型。同样地，我们必须实现一个.processElement()抽象方法，用来处理流中的每一个数据；另外还有一个非抽象方法.onTimer()，用来定义定时器触发时的回调操作。
代码如下：
```java
public class KeyedProcessTimerDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator sensorDS = env
.socketTextStream("hadoop102", 7777)
.map(new WaterSensorMapFunction())
.assignTimestampsAndWatermarks(
WatermarkStrategy
.