Flink1.17实战教程（第一篇：概念、部署、架构）_flink1.17项目开发

系列文章目录

Flink1.17实战教程（第一篇：概念、部署、架构）
Flink1.17实战教程（第二篇：DataStream API）
Flink1.17实战教程（第三篇：时间和窗口）
Flink1.17实战教程（第四篇：处理函数）
Flink1.17实战教程（第五篇：状态管理）
Flink1.17实战教程（第六篇：容错机制）
Flink1.17实战教程（第七篇：Flink SQL）

---

---

1. Flink 概述

1.1 Flink 是什么

Flink官网
Flimk核心目标，是“数据流上的有状态计算”（Stateful Computati ons over Data Streams）
具体说明：Apache Flink是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。

有界流和无界流：

（1）无界流数据

• 有定义流的开始，但没有定义流的结束；
• 它们会无休止的产生数据；
• 无界流的数据必须持续处理，即数据被摄取后需要立刻处理我们不能等到所有数据都到达再处理，因为输入是无限的。

（2）有界流数据

• 有定义流的开始，也有定义流的结束；
• 有界流可以在摄取所有数据后再进行计算；
• 有界流所有数据可以被排序，所以并不需要有序摄取；
• 有界流处理通常被称为批处理。

有状态流处理

把流处理需要的额外数据保存成一个“状态”，然后针对这条数据进行处理，并且更新状态。这就是所谓的“有状态的流处理”。

• 状态在内存中：优点，速度快；缺点，可靠性差
• 状态在分布式系统中：优点，可靠性高；缺点，速度慢

Flink的发展历史

Flink起源于一个叫作Stratosphere的项目，它是由3所地处柏林的大学和欧洲其他一些大学在2010~2014年共同进行的研究项目，由柏林理工大学的教授沃克尔·马尔科(volker Markl)领衔开发。2014年4月，Stratosphere的代码被复制并捐赠给了Apache软件基金会，Flnk就是在此基础上被重新设计出来的。
在德语中，“flink”一词表示“快速、灵巧”
项目的logo是一只彩色的松鼠。

• 2014年8月，Flink第一个版本0.6正式发布，与此同时Fk的几位核心开发者创办Data Artisans公司
• 2014年12月，Flink项目完成孵化
• 2015年4月，Flink发布了里程碑式的重要版本0.9.0
• 2019年1月，长期对Flink投入研发的阿里巴巴，以9000万欧元的价格收购了Data Artisans 公司
• 2019年8月，阿里巴巴将内部版本Blink开源，合并入Flink 1.9.0版本

1.2 Flink 特点

我们处理数据的目标是：低延迟、高吞吐、结果的准确性和良好的容错性。

Flink主要特点如下：

• 高吞吐和低延迟。每秒处理数百万个事件，毫秒级延迟
• 结果的准确性。Flink提供了事件时间 (event-time)和处理时间 (processing time)语义。对于乱序事件流，事件时间语义仍然能提供一致且准确的结果。
• 精确一次(exactly - once)的状态一致性保证。
• 可以连接到最常用的外部系统，如Kafka、Hive、JDBC、HDFS、Redis等。
• 高可用。本身高可用的设置，加上与K8s，YARN和Mesos的紧密集成，再加上从故障中快速恢复和动态扩展任务的能力，Flink能做到以极少的停机时间7X24全天候运行。

1.3 Flink vs SparkStreaming

Spark以批处理为根本

• Spark数据模型：Spark 采用RDD 模型，Spark Streaming 的 DStream 实际上也就是一组组小批数据 RDD 的集台
• Spark运行时架构：Spark 是批计算，将 DAG 分为不同的 stage，一个完成后才可以计算下一个

Flink以流处理为根本

• Flink数据模型：Flink基本数据模型是数据流，以及事件（Event）序列
• Flink运行时架构：Flink是标准的流执行模式，一个事件在一个节点处理完后可以直接发往下一个节点进行处理
  

1.4 Flink 的应用场景

Flink在国内各个企业中大量使用。一些行业中的典型应用有：
（1）电商和市场营销
举例：实时数据报表、广告投放、实时推荐
（2）物联网（IOT）
举例：传感器实时数据采集和显示、实时报警，交通运输业
（3）物流和服务业
举例：订单状态实时更新、通知信息推送
（4）银行和金融业
举例：实时结算和通知推送，实时检测异常行为

1.5 Flink 分层API

• 越顶层越抽象，表达含义越简明，使用越方便
• 越底层越具体，表达能力越丰富，使用越灵活
  

有状态流处理：通过底层API（处理函数），对最原始数据加工处理。底层API与DataStream API相集成，可以处理复杂的计算。

DataStream API(流处理)和DataSet API(批处理)封装了底层处理函数，提供了通用的模块，比如转换(transformations，包括map、flatmap等)，连接 (joins)，聚合 (aggregations)，窗口 (windows) 操作等，注意：Flnk1.12以后，DataSteam API已经实现真正的流批一体，所以DataSet API已经过时。

Table API是以表为中心的声明式编程，其中表可能会动态变化。Table API遵循关系模型：表有二维数据结构，类似于关系数据库中的表；同时API提供可比较的操作，例如select、 project、join、group-by、aggregate等。我们可以在表与 DataStream / DataSet之间无缝切换，以允许程序将 Table API与 DataStream 以及 DataSet 混合使用。

SQL这一层在语法与表达能力上与 Table API类似，但是是以SQL查询表达式的形式表现程序。SQL抽象与Table API交互密切，同时SOL查询可以直接在Table API定义的表上执行。

2. Flink 快速上手

2.1 创建项目

在准备好所有的开发环境之后，我们就可以开始开发自己的第一个Flink程序了。首先我们要做的，就是在IDEA中搭建一个Flink项目的骨架。我们会使用Java项目中常见的Maven来进行依赖管理。

2.1.1 创建工程

（1）打开IntelliJ IDEA，创建一个Maven工程。

（2）将这个Maven工程命名为FlinkTutorial。

（3）选定这个Maven工程所在存储路径，并点击Finish，Maven工程即创建成功。

2.1.2 添加项目依赖

在项目的pom文件中，添加Flink的依赖，包括flink-java、flink-streaming-java，以及flink-clients（客户端，也可以省略）。

<properties>
        <flink.version>1.17.0</flink.version>
</properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
        </dependency>

     <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
     </dependency>
</dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

2.2 WordCount代码编写

需求：统计一段文字中，每个单词出现的频次。
环境准备：在src/main/java目录下，新建一个包，命名为com.atguigu.wc。

2.2.1 批处理

批处理基本思路：先逐行读入文件数据，然后将每一行文字拆分成单词；接着按照单词分组，统计每组数据的个数，就是对应单词的频次。

数据准备
（1）在工程根目录下新建一个input文件夹，并在下面创建文本文件words.txt
（2）在words.txt中输入一些文字，例如：

hello flink
hello world
hello java
1
2
3

代码编写
（1）在com.atguigu.wc包下新建Java类BatchWordCount，在静态main方法中编写代码。具体代码实现如下：

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<String> lineDS = env.readTextFile("input/words.txt");
        
        // 3. 转换数据格式
        FlatMapOperator<String, Tuple2<String, Long>> wordAndOne = lineDS.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {

            @Override
            public void flatMap(String line, Collector<Tuple2<String, Long>> out) throws Exception {

                String[] words = line.split(" ");

                for (String word : words) {
                    out.collect(Tuple2.of(word,1L));
                }
            }
        });

        // 4. 按照 word 进行分组
        UnsortedGrouping<Tuple2<String, Long>> wordAndOneUG = wordAndOne.groupBy(0);
        
        // 5. 分组内聚合统计
        AggregateOperator<Tuple2<String, Long>> sum = wordAndOneUG.sum(1);

        // 6. 打印结果
        sum.print();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

（2）输出

(flink,1)
(world,1)
(hello,3)
(java,1)
1
2
3
4

需要注意的是，这种代码的实现方式，是基于DataSet API的，也就是我们对数据的处理转换，是看作数据集来进行操作的。事实上Flink本身是流批统一的处理架构，批量的数据集本质上也是流，没有必要用两套不同的API来实现。所以从Flink 1.12开始，官方推荐的做法是直接使用DataStream API，在提交任务时通过将执行模式设为BATCH来进行批处理：

$ bin/flink run -Dexecution.runtime-mode=BATCH BatchWordCount.jar
1

这样，DataSet API就没什么用了，在实际应用中我们只要维护一套DataStream API就可以。这里只是为了方便大家理解，我们依然用DataSet API做了批处理的实现。

2.2.2 流处理

对于Flink而言，流才是整个处理逻辑的底层核心，所以流批统一之后的DataStream API更加强大，可以直接处理批处理和流处理的所有场景。
下面我们就针对不同类型的输入数据源，用具体的代码来实现流处理。

读取文件
我们同样试图读取文档words.txt中的数据，并统计每个单词出现的频次。整体思路与之前的批处理非常类似，代码模式也基本一致。
在com.atguigu.wc包下新建Java类StreamWordCount，在静态main方法中编写代码。具体代码实现如下：

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<String> lineStream = env.readTextFile("input/words.txt");
        
        // 3. 转换、分组、求和，得到统计结果
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = lineStream.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Long>> 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();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

输出：

3> (java,1)
5> (hello,1)
5> (hello,2)
5> (hello,3)
13> (flink,1)
9> (world,1)
1
2
3
4
5
6

主要观察与批处理程序BatchWordCount的不同：

• 创建执行环境的不同，流处理程序使用的是StreamExecutionEnvironment。
• 转换处理之后，得到的数据对象类型不同。
• 分组操作调用的是keyBy方法，可以传入一个匿名函数作为键选择器（KeySelector），指定当前分组的key是什么。
• 代码末尾需要调用env的execute方法，开始执行任务。

读取socket文本流
在实际的生产环境中，真正的数据流其实是无界的，有开始却没有结束，这就要求我们需要持续地处理捕获的数据。为了模拟这种场景，可以监听socket端口，然后向该端口不断的发送数据。

（1）将StreamWordCount代码中读取文件数据的readTextFile方法，替换成读取socket文本流的方法socketTextStream。具体代码实现如下：

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<String> lineStream = env.socketTextStream("hadoop102", 7777);
        
        // 3. 转换、分组、求和，得到统计结果
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = lineStream.flatMap((String line, Collector<Tuple2<String, Long>> 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();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

（2）在Linux环境的主机hadoop102上，执行下列命令，发送数据进行测试

[atguigu@hadoop102 ~]$ nc -lk 7777
1

注意：要先启动端口，后启动StreamWordCount程序，否则会报超时连接异常。

（3）启动StreamWordCount程序
我们会发现程序启动之后没有任何输出、也不会退出。这是正常的，因为Flink的流处理是事件驱动的，当前程序会一直处于监听状态，只有接收到数据才会执行任务、输出统计结果。

（4）从hadoop102发送数据
①在hadoop102主机中，输入“hello flink”，输出如下内容

13> (flink,1)
5> (hello,1)
1
2

②再输入“hello world”，输出如下内容

2> (world,1)
5> (hello,2)
1
2

说明：
Flink还具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息，从而获得对应的序列化器和反序列化器。但是，由于Java中泛型擦除的存在，在某些特殊情况下（比如Lambda表达式中），自动提取的信息是不够精细的——只告诉Flink当前的元素由“船头、船身、船尾”构成，根本无法重建出“大船”的模样；这时就需要显式地提供类型信息，才能使应用程序正常工作或提高其性能。
因为对于flatMap里传入的Lambda表达式，系统只能推断出返回的是Tuple2类型，而无法得到Tuple2<String, Long>。只有显式地告诉系统当前的返回类型，才能正确地解析出完整数据。

3. Flink 部署

3.1 集群角色

Flink提交作业和执行任务，需要几个关键组件：

• 客户端(Client)：代码由客户端获取并做转换，之后提交给JobManger
• JobManager就是Flink集群里的“管事人”，对作业进行中央调度管理，而它获取到要执行的作业后，会进一步处理转换，然后分发任务给众多的TaskManager。
• TaskManager，就是真正“干活的人”，数据的处理操作都是它们来做的。
  
  注意：Flink是一个非常灵活的处理框架，它支持多种不同的部署场景，还可以和不同的资源管理平台方便地集成。

3.2 Flink 集群搭建

3.2.1 集群启动

0）集群规划

具体安装部署步骤如下：

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路径上。

[atguigu@hadoop102 software]$ tar -zxvf flink-1.17.0-bin-scala_2.12.tgz -C /opt/module/
1

2）修改集群配置
（1）进入conf路径，修改flink-conf.yaml文件，指定hadoop102节点服务器为JobManager

[atguigu@hadoop102 conf]$ vim flink-conf.yaml
1

修改如下内容：

# 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
1
2
3
4
5
6
7
8

（2）修改workers文件，指定hadoop102、hadoop103和hadoop104为TaskManager

[atguigu@hadoop102 conf]$ vim workers
1

修改如下内容：

hadoop102
hadoop103
hadoop104
1
2
3

（3）修改masters文件

[atguigu@hadoop102 conf]$ vim masters
1

修改如下内容：

hadoop102:8081
1

（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。优先级低于代码中进行的并行度配置和任务提交时使用参数指定的并行度数量。

3）分发安装目录
（1）配置修改完毕后，将Flink安装目录发给另外两个节点服务器。

[atguigu@hadoop102 module]$ xsync flink-1.17.0/
1

（2）修改hadoop103的 taskmanager.host

[atguigu@hadoop103 conf]$ vim flink-conf.yaml
1

修改如下内容：

# TaskManager节点地址.需要配置为当前机器名
taskmanager.host: hadoop103
1
2

（3）修改hadoop104的 taskmanager.host

[atguigu@hadoop104 conf]$ vim flink-conf.yaml
1

修改如下内容：

# TaskManager节点地址.需要配置为当前机器名
taskmanager.host: hadoop104
1
2

4）启动集群
（1）在hadoop102节点服务器上执行start-cluster.sh启动Flink集群：

[atguigu@hadoop102 flink-1.17.0]$ bin/start-cluster.sh
1

（2）查看进程情况：

[atguigu@hadoop102 flink-1.17.0]$ jpsall 
=============== hadoop102 ===============
4453 StandaloneSessionClusterEntrypoint
4458 TaskManagerRunner
4533 Jps
=============== hadoop103 ===============
2872 TaskManagerRunner
2941 Jps
=============== hadoop104 ===============
2948 Jps
2876 TaskManagerRunner
1
2
3
4
5
6
7
8
9
10
11

5）访问Web UI
启动成功后，同样可以访问http://hadoop102:8081对flink集群和任务进行监控管理。

这里可以明显看到，当前集群的TaskManager数量为3；由于默认每个TaskManager的Slot数量为1，所以总Slot数和可用Slot数都为3。

3.2.2 向集群提交作业

1）环境准备
在hadoop102中执行以下命令启动netcat。

[atguigu@hadoop102 flink-1.17.0]$ nc -lk 7777
1

2）程序打包
（1）在我们编写的Flink入门程序的pom.xml文件中添加打包插件的配置，具体如下：

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>3.2.4</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                    <configuration>
                        <artifactSet>
                            <excludes>
                            	<exclude>com.google.code.findbugs:jsr305</exclude>
                                <exclude>org.slf4j:*</exclude>
                                <exclude>log4j:*</exclude>
                            </excludes>
                        </artifactSet>
                        <filters>
                            <filter>
                                <!-- Do not copy the signatures in the META-INF folder.
                                Otherwise, this might cause SecurityExceptions when using the JAR. -->
                                <artifact>*:*</artifact>
                                <excludes>
                                    <exclude>META-INF/*.SF</exclude>
                                    <exclude>META-INF/*.DSA</exclude>
                                    <exclude>META-INF/*.RSA</exclude>
                                </excludes>
                            </filter>
                        </filters>
                        <transformers combine.children="append">
                            <transformer
                                    implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer">
                            </transformer>
                        </transformers>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

（2）插件配置完毕后，可以使用IDEA的Maven工具执行package命令，出现如下提示即表示打包成功。

-------------------------------------------------------------------
[INFO] BUILD SUCCESS
-------------------------------------------------------------------
1
2
3

打包完成后，在target目录下即可找到所需JAR包，JAR包会有两个，FlinkTutorial-1.0-SNAPSHOT.jar和FlinkTutorial-1.0-SNAPSHOT-jar-with-dependencies.jar，因为集群中已经具备任务运行所需的所有依赖，所以建议使用FlinkTutorial-1.0-SNAPSHOT.jar。

3）在Web UI上提交作业
（1）任务打包完成后，我们打开Flink的WEB UI页面，在右侧导航栏点击“Submit New Job”，然后点击按钮“+ Add New”，选择要上传运行的JAR包，如下图所示。

JAR包上传完成，如下图所示：

（2）点击该JAR包，出现任务配置页面，进行相应配置。
主要配置程序入口主类的全类名，任务运行的并行度，任务运行所需的配置参数和保存点路径等，如下图所示，配置完成后，即可点击按钮“Submit”，将任务提交到集群运行。

（3）任务提交成功之后，可点击左侧导航栏的“Running Jobs”查看程序运行列表情况。

（4）测试
①在socket端口中输入hello

[atguigu@hadoop102 flink-1.17.0]$ nc -lk 7777
hello
1
2

②先点击Task Manager，然后点击右侧的192.168.10.104服务器节点

③点击Stdout，就可以看到hello单词的统计

注意：如果hadoop104节点没有统计单词数据，可以去其他TaskManager节点查看。
④点击该任务，可以查看任务运行的具体情况，也可以通过点击“Cancel Job”结束任务运行。

4）命令行提交作业

除了通过WEB UI界面提交任务之外，也可以直接通过命令行来提交任务。这里为方便起见，我们可以先把jar包直接上传到目录flink-1.17.0下
（1）首先需要启动集群。

[atguigu@hadoop102 flink-1.17.0]$ bin/start-cluster.sh
1

（2）在hadoop102中执行以下命令启动netcat。

[atguigu@hadoop102 flink-1.17.0]$ nc -lk 7777
1

（3）将flink程序运行jar包上传到/opt/module/flink-1.17.0路径。
（4）进入到flink的安装路径下，在命令行使用flink run命令提交作业。

[atguigu@hadoop102 flink-1.17.0]$ bin/flink run -m hadoop102:8081 -c com.atguigu.wc.SocketStreamWordCount ./FlinkTutorial-1.0-SNAPSHOT.jar
1

这里的参数 -m指定了提交到的JobManager，-c指定了入口类。
（5）在浏览器中打开Web UI，http://hadoop102:8081查看应用执行情况。
用netcat输入数据，可以在TaskManager的标准输出（Stdout）看到对应的统计结果。

（6）在/opt/module/flink-1.17.0/log路径中，可以查看TaskManager节点。

[atguigu@hadoop102 log]$ cat flink-atguigu-standalonesession-0-hadoop102.out

(hello,1)
(hello,2)
(flink,1)
(hello,3)
(scala,1)
1
2
3
4
5
6
7

3.3 部署模式

在一些应用场景中，对于集群资源分配和占用的方式，可能会有特定的需求。Flink为各种场景提供了不同的部署模式，主要有以下三种：会话模式（Session Mode）、单作业模式（Per-Job Mode）、应用模式（Application Mode）。
它们的区别主要在于：集群的生命周期以及资源的分配方式；以及应用的main方法到底在哪里执行——客户端（Client）还是JobManager。

3.3.1 会话模式（Session Mode）

会话模式其实最符合常规思维。我们需要先启动一个集群，保持一个会话，在这个会话中通过客户端提交作业。集群启动时所有资源就都已经确定，所以所有提交的作业会竞争集群中的资源。

会话模式比较适合于单个规模小、执行时间短的大量作业。

3.3.2 单作业模式（Per-Job Mode）

会话模式因为资源共享会导致很多问题，所以为更好地隔离资源，我们可以考虑为每个提交的作业启动一个集群，这就是所谓的单作业《Per-Job) 模式。

作业完成后，集群就会关闭，所有资源也会释放
这些特性使得单作业模式在生产环境运行更加稳定，所以是实际应用的首选模式
需要注意的是，Flink本身无法直接这样运行，所以单作业模式一般需要借助一些资源管理框架来启动集群，比如YARN、Kubernetes (K8S)

3.3.3 应用模式（Application Mode）

前面提到的两种模式下，应用代码都是在客户端上执行，然后由客户端提交给JobMareget的。但是这种方式客户端需要占用大重网络带宽，去下载依和把二进制教发送给Job Manager; 加上很多况下我们提文作业用的是同一个客户端，就会加重客户端所在节点的资源消耗
所以解决办法就是，我们不要客户端了，直接把应用提交到JcbManeer上运行。而这也就代表着，我们需要为每一个提交的应用单独启动一个JobMareger，也就是创建一个集群。这个JobManager只为执行这一个应用而存在执行结束之后JobMareger也就关闭了，这就是所谓的应用模式

应用模式与单作业模式，都是提交作业之后才创建案群，单作业模式是通过客户端来提交的，客户端解析出的每一个作业对应一个集群;而应用模式下，是直接由JobManager执行应用程序的。

这里我们所讲到的部署模式，相对是比较抽象的概念。实际应用时，一般需要和资源管理平台结合起来，选择特定的模式来分配资源、部署应用。接下来，我们就针对不同的资源提供者的场景，具体介绍Flink的部署方式。

3.4 Standalone运行模式（了解）

独立模式是独立运行的，不依赖任何外部的资源管理平台；当然独立也是有代价的：如果资源不足，或者出现故障，没有自动扩展或重分配资源的保证，必须手动处理。所以独立模式一般只用在开发测试或作业非常少的场景下。

3.4.1 会话模式部署

提前启动集群，并通过Web页面客户端提交任务（可以多个任务，但是集群资源固定）。

3.4.2 单作业模式部署

Flink的Standalone集群并不支持单作业模式部署。因为单作业模式需要借助一些资源管理平台。

3.4.3 应用模式部署

应用模式下不会提前创建集群，所以不能调用start-cluster.sh脚本。我们可以使用同样在bin目录下的standalone-job.sh来创建一个JobManager。

具体步骤如下：
（0）环境准备。在hadoop102中执行以下命令启动netcat。

[atguigu@hadoop102 flink-1.17.0]$ nc -lk 7777
1

（1）进入到Flink的安装路径下，将应用程序的jar包放到lib/目录下。

[atguigu@hadoop102 flink-1.17.0]$ mv FlinkTutorial-1.0-SNAPSHOT.jar lib/
1

（2）执行以下命令，启动JobManager。

[atguigu@hadoop102 flink-1.17.0]$ bin/standalone-job.sh start --job-classname com.atguigu.wc.SocketStreamWordCount
1

这里我们直接指定作业入口类，脚本会到lib目录扫描所有的jar包。
（3）同样是使用bin目录下的脚本，启动TaskManager。

[atguigu@hadoop102 flink-1.17.0]$ bin/taskmanager.sh start
1

（4）在hadoop102上模拟发送单词数据。

[atguigu@hadoop102 ~]$ nc -lk 7777
hello
1
2

（5）在hadoop102:8081地址中观察输出数据

（6）如果希望停掉集群，同样可以使用脚本，命令如下。

[atguigu@hadoop102 flink-1.17.0]$ bin/taskmanager.sh stop
[atguigu@hadoop102 flink-1.17.0]$ bin/standalone-job.sh stop
1
2

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）配置环境变量，增加环境变量配置如下：

$ 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`
1
2
3
4
5
6

（2）启动Hadoop集群，包括HDFS和YARN。

[atguigu@hadoop102 hadoop-3.3.4]$ start-dfs.sh
[atguigu@hadoop103 hadoop-3.3.4]$ start-yarn.sh
1
2

（3）在hadoop102中执行以下命令启动netcat。

[atguigu@hadoop102 flink-1.17.0]$ nc -lk 7777
1

3.5.2 会话模式部署

YARN的会话模式与独立集群略有不同，需要首先申请一个YARN会话（YARN Session）来启动Flink集群。具体步骤如下：
1）启动集群
（1）启动Hadoop集群（HDFS、YARN）。
（2）执行脚本命令向YARN集群申请资源，开启一个YARN会话，启动Flink集群。
[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或者命令行两种方式提交作业。

2）提交作业
（1）通过Web UI提交作业
这种方式比较简单，与上文所述Standalone部署模式基本相同。

（2）通过命令行提交作业
① 将FlinkTutorial-1.0-SNAPSHOT.jar任务上传至集群。
② 执行以下命令将该任务提交到已经开启的Yarn-Session中运行。

[atguigu@hadoop102 flink-1.17.0]$ bin/flink run
-c com.atguigu.wc.SocketStreamWordCount FlinkTutorial-1.0-SNAPSHOT.jar
1
2

客户端可以自行确定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）执行命令提交作业。

[atguigu@hadoop102 flink-1.17.0]$ bin/flink run -d -t yarn-per-job -c com.atguigu.wc.SocketStreamWordCount FlinkTutorial-1.0-SNAPSHOT.jar
1

注意：如果启动过程中报如下异常。

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
1
2

解决办法：在flink的/opt/module/flink-1.17.0/conf/flink-conf.yaml配置文件中设置

[atguigu@hadoop102 conf]$ vim flink-conf.yaml
classloader.check-leaked-classloader: false
1
2

（2）在YARN的ResourceManager界面查看执行情况。

点击可以打开Flink Web UI页面进行监控，如下图所示：

（3）可以使用命令行查看或取消作业，命令如下。

[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 <jobId>
1
2
3

这里的application_XXXX_YY是当前应用的ID，是作业的ID。注意如果取消作业，整个Flink集群也会停掉。

3.5.4 应用模式部署

应用模式同样非常简单，与单作业模式类似，直接执行flink run-application命令即可。

1）命令行提交

（1）执行命令提交作业。

[atguigu@hadoop102 flink-1.17.0]$ bin/flink run-application -t yarn-application -c com.atguigu.wc.SocketStreamWordCount FlinkTutorial-1.0-SNAPSHOT.jar 
1

（2）在命令行中查看或取消作业。

[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 <jobId>
1
2
3

2）上传HDFS提交

可以通过yarn.provided.lib.dirs配置选项指定位置，将flink的依赖上传到远程。

（1）上传flink的lib和plugins到HDFS上

[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
1
2
3

（2）上传自己的jar包到HDFS

[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
1
2

（3）提交作业

[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
1

这种方式下，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）创建存储目录

hadoop fs -mkdir -p /logs/flink-job
1

2）在 flink-config.yaml中添加如下配置

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
1
2
3
4
5

3）启动历史服务器

bin/historyserver.sh start
1

4）停止历史服务器

bin/historyserver.sh stop
1

5）在浏览器地址栏输入：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()方法，来设置当前算子的并行度：

stream.map(word -> Tuple2.of(word, 1L)).setParallelism(2);
1

这种方式设置的并行度，只针对当前算子有效。
另外，我们也可以直接调用执行环境的setParallelism()方法，全局设定并行度：

env.setParallelism(2);
1

这样代码中所有算子，默认的并行度就都为2了。我们一般不会在程序中设置全局并行度，因为如果在程序中对全局并行度进行硬编码，会导致无法动态扩容。
这里要注意的是，由于keyBy不是算子，所以无法对keyBy设置并行度。

（2）提交应用时设置
在使用flink run命令提交应用时，可以增加-p参数来指定当前应用程序执行的并行度，它的作用类似于执行环境的全局设置：

bin/flink run –p 2 –c com.atguigu.wc.SocketStreamWordCount  ./FlinkTutorial-1.0-SNAPSHOT.jar
1

如果我们直接在Web UI上提交作业，也可以在对应输入框中直接添加并行度。

（3）配置文件中设置
我们还可以直接在集群的配置文件flink-conf.yaml中直接更改默认并行度：

parallelism.default: 2
1

这个设置对于整个集群上提交的所有作业有效，初始值为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默认会按照算子链的原则进行链接合并，如果我们想要禁止合并或者自行定义，也可以在代码中对算子做一些特定的设置：

// 禁用算子链
.map(word -> Tuple2.of(word, 1L)).disableChaining();

// 从当前算子开始新链
.map(word -> Tuple2.of(word, 1L)).startNewChain()
1
2
3
4
5

4.2.3 任务槽（Task Slots）

1）任务槽（Task Slots）

Flink中每一个TaskManager都是一个JVM进程，它可以启动多个独立的线程，来并行执行多个子任务（subtask）。
很显然，TaskManager的计算资源是有限的，并行的任务越多，每个线程的资源就会越少。那一个TaskManager到底能并行处理多少个任务呢？为了控制并发量，我们需要在TaskManager上对每个任务运行所占用的资源做出明确的划分，这就是所谓的任务槽（task slots）。
每个任务槽（task slot）其实表示了TaskManager拥有计算资源的一个固定大小的子集。这些资源就是用来独立执行一个子任务的。

假如一个TaskManager有三个slot，那么它会将管理的内存平均分成三份，每个slot独自占据一份。这样一来，我们在slot上执行一个子任务时，相当于划定了一块内存“专款专用”，就不需要跟来自其他作业的任务去竞争内存资源了。
所以现在我们只要2个TaskManager，就可以并行处理分配好的5个任务了

2）任务槽数量的设置

在Flink的/opt/module/flink-1.17.0/conf/flink-conf.yaml配置文件中，可以设置TaskManager的slot数量，默认是1个slot。

taskmanager.numberOfTaskSlots: 8
1

需要注意的是，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共享组”手动指定：

.map(word -> Tuple2.of(word, 1L)).slotSharingGroup("1");
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可以合并算子链，于是最终有三个任务节点。

任务槽和并行度的关系（1）

任务槽和并行度的关系（2）

任务槽和并行度的关系（3）

任务槽和并行度的关系（4）

通过这个例子也可以明确地看到，整个流处理程序的并行度，就应该是所有算子并行度中最大的那个，这代表了运行程序需要的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应用模式作业提交流程