Flink运行时架构

1. Standalone会话模式系统架构

Alt text

作业管理器(JobManager): JobManager是一个Flink集群中任务管理和调度的核心，是控制应用执行的主进程。也就是说，每个应用都被唯一对应的JobManager所控制执行。它相当于Spark集群中的ApplicationMaster概念。JobManager又包含3个不同的组件:
- JobMaster: JobMaster是JobManager中最核心的组件，负责处理单独的作业(Job)。相当于Spark中的Driver概念。所以JobMaster和具体的Job是一一对应的，多个Job可以同时运行在一个Flink集群中, 每个Job都有一个自己的JobMaster。需要注意在早期版本的Flink中，没有JobMaster的概念；而JobManager的概念范围较小，实际指的就是现在所说的JobMaster。在作业提交时，JobMaster会先接收到要执行的应用。JobMaster会把JobGraph转换成一个物理层面的数据流图，这个图被叫作"执行图"(ExecutionGraph)，它包含了所有可以并发执行的任务。JobMaster会向资源管理器(ResourceManager)发出请求，申请执行任务必要的资源。一旦它获取到了足够的资源，就会将执行图分发到真正运行它们的TaskManager上。而在运行过程中，JobMaster会负责所有需要中央协调的操作，比如说检查点(checkpoints)的协调。
- 资源管理器(ResourceManager): ResourceManager主要负责资源的分配和管理，在Flink集群中只有一个。所谓"资源"，主要是指TaskManager的任务槽(task slots)。任务槽就是Flink集群中的资源调配单元，包含了机器用来执行计算的一组CPU和内存资源。每一个任务(Task)都需要分配到一个slot上执行。这里注意要把Flink内置的ResourceManager和其他资源管理平台(比如YARN)的ResourceManager区分开。
- 分发器(Dispatcher): Dispatcher主要负责提供一个REST接口，用来提交应用，并且负责为每一个新提交的作业启动一个新的JobMaster组件。Dispatcher也会启动一个Web UI，用来方便地展示和监控作业执行的信息。Dispatcher在架构中并不是必需的，在不同的部署模式下可能会被忽略掉。
任务管理器(TaskManager): TaskManager是Flink中的工作进程，数据流的具体计算就是它来做的。Flink集群中必须至少有一个TaskManager；每一个TaskManager都包含了一定数量的任务槽(task slots)。Slot是资源调度的最小单位，slot的数量限制了TaskManager能够并行处理的任务数量。启动之后，TaskManager会向资源管理器注册它的slots；收到资源管理器的指令后，TaskManager就会将一个或者多个槽位提供给JobMaster调用，JobMaster就可以分配任务来执行了。在执行过程中，TaskManager可以缓冲数据，还可以跟其他运行同一应用的TaskManager交换数据。

2. 并行度(Parallelism)

2.1 并行子任务和并行度

当要处理的数据量非常大时，我们可以把一个算子操作，"复制"多份到多个节点，数据来了之后就可以到其中任意一个执行。这样一来，一个算子任务就被拆分成了多个并行的"子任务"(subtasks)，再将它们分发到不同节点，就真正实现了并行计算。
在Flink执行过程中，每一个算子(operator)可以包含一个或多个子任务(operator subtask)，这些子任务在不同的线程、不同的物理机或不同的容器中完全独立地执行。
Alt text 一个特定算子的子任务(subtask)的个数被称之为其并行度(parallelism)。这样，包含并行子任务的数据流，就是并行数据流，它需要多个分区(stream partition)来分配并行任务。一般情况下，一个流程序的并行度，可以认为就是其所有算子中最大的并行度。一个程序中，不同的算子可能具有不同的并行度。例如：如上图所示，当前数据流中有source、map、window、sink四个算子，其中sink算子的并行度为1，其他算子的并行度都为2。所以这段流处理程序的并行度就是2。

2.2 并行度的设置

在Flink中，可以用不同的方法来设置并行度，它们的有效范围和优先级别也是不同的。

代码中设置
我们在代码中，可以很简单地在算子后跟着调用setParallelism()方法，来设置当前算子的并行度：

SocketStreamWordCount2.scalapom.xml

scala

def main(args: Array[String]): Unit = {
    // 1. 创建流式执行环境, createLocalEnvironmentWithWebUI---主要用于开发测试，在本地IDEA中也能看到webui
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration())
    // webui还需要一个依赖 flink-runtime-web
    
    // 2. 读取socket文本流数据
    // 可以从外部传参，传参格式为--host 192.168.101.102  --port 7777
    val parameters = ParameterTool.fromArgs(args)
    val host = parameters.get("host", "hadoop102")
    val port = parameters.getInt("port", 7777)

    val lineStream = env.socketTextStream(host, port)
    // 3. 转换、分组、求和，得到统计结果
    val sum: DataStream[(String, Int)] = lineStream
        .flatMap(_.split(" ")).setParallelism(2)
        .map((_, 1))
        .keyBy(_._1)
        .sum(1)
    sum.print()
    env.execute()
}

xml

<properties>
    <maven.compiler.source>8</maven.compiler.source>
    <maven.compiler.target>8</maven.compiler.target>
    <maven.compiler.compilerVersion>1.8</maven.compiler.compilerVersion>
    <target.java.version>1.8</target.java.version>
    <flink.version>1.17.2</flink.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-runtime-web</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
</dependencies>

这种方式设置的并行度，只针对当前算子有效, 访问localhost:8081即可访问测试的webui。 Alt text 另外，我们也可以直接调用执行环境的setParallelism()方法，全局设定并行度：

java

env.setParallelism(2)  // 设置全局并行度

这样代码中所有算子，默认的并行度就都为2了。我们一般不会在程序中设置全局并行度，因为如果在程序中对全局并行度进行硬编码，会导致无法动态扩容。这里要注意的是，由于keyBy不是算子，所以无法对keyBy设置并行度。另外算子中设置并行度优先级高于全局设置并行度。
2. 提交应用时设置
在使用flink run命令提交应用时，可以增加-p参数来指定当前应用程序执行的并行度，它的作用类似于执行环境的全局设置：

bin/flink run –p 2 –c com.rocket.flink.SocketStreamWordCount ./flink-demo-1.0-SNAPSHOT.jar

如果我们直接在Web UI上提交作业，也可以在对应输入框中直接添加并行度。
3. 配置文件中设置
我们还可以直接在集群的配置文件flink-conf.yaml中直接更改默认并行度：

yml

parallelism.default: 2

这个设置对于整个集群上提交的所有作业有效，初始值为1。无论在代码中设置、还是提交时的-p参数，都不是必须的；所以在没有指定并行度的时候，就会采用配置文件中的集群默认并行度。在开发环境中，没有配置文件，默认并行度就是当前机器的CPU核心数。

3. 算子链(Operator Chain)

3.1 算子间的数据传输

Alt text 一个数据流在算子之间传输数据的形式可以是一对一(one-to-one)的直通(forwarding)模式，也可以是打乱的重分区(redistributing)模式，具体是哪一种形式，取决于算子的种类。

3.2 一对一(One-to-one，forwarding)

这种模式下，数据流维护着分区以及元素的顺序。比如图中的source和map算子，source算子读取数据之后，可以直接发送给map算子做处理，它们之间不需要重新分区，也不需要调整数据的顺序。这就意味着map 算子的子任务，看到的元素个数和顺序跟source 算子的子任务产生的完全一样，保证着"一对一"的关系。map、filter、flatMap等算子都是这种one-to-one的对应关系。这种关系类似于Spark中的窄依赖。

3.3 重分区(Redistributing)

在这种模式下，数据流的分区会发生改变。比如图中的map和后面的keyBy/window算子之间，以及keyBy/window算子和Sink算子之间，都是这样的关系。
每一个算子的子任务，会根据数据传输的策略，把数据发送到不同的下游目标任务。这些传输方式都会引起重分区的过程，这一过程类似于Spark中的shuffle。

3.4 合并算子链

在Flink中，并行度相同的一对一(one to one)算子操作，可以直接链接在一起形成一个"大"的任务(task)，这样原来的算子就成为了真正任务里的一部分，如下图所示。每个task会被一个线程执行。这样的技术被称为"算子链"(Operator Chain)。 Alt text 上图中Source和map之间满足了算子链的要求，所以可以直接合并在一起，形成了一个任务；因为并行度为2，所以合并后的任务也有两个并行子任务。这样，这个数据流图所表示的作业最终会有5个任务，由5个线程并行执行。
将算子链接成task是非常有效的优化：可以减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。
Flink默认会按照算子链的原则进行链接合并，如果我们想要禁止合并或者自行定义，也可以在代码中对算子做一些特定的设置：

java

public class SocketStreamWordCount3 {
    public static void main(String[] args) throws Exception {
        // 1. 创建流式执行环境, createLocalEnvironmentWithWebUI---主要用于开发测试，在本地IDEA中也能看到webui
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        // webui还需要一个依赖 flink-runtime-web
        // 设置全局禁用合并算子链
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        env.disableOperatorChaining();
        // 2. 读取socket文本流数据
        // 可以从外部传参，传参格式为--host 192.168.101.102  --port 7777
        String host = parameterTool.get("host", "hadoop102");
        Integer port = parameterTool.getInt("port", 7777);

        DataStreamSource<String> lineStream = env.socketTextStream(host, port);
        // 3. 转换、分组、求和，得到统计结果
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumStream = lineStream
                .flatMap((value, out) -> {
                    String[] strArray = value.split(" ");
                    for (String str : strArray) {
                        out.collect(str);
                    }
                }, TypeInformation.of(String.class)).setParallelism(2)
                // 禁用算子链
                .map(value -> Tuple2.of(value, 1), TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {})).disableChaining()
                .keyBy(item -> item.f0)
                // 从当前算子开始新链
                .sum(1).startNewChain();
        sumStream.print();
        env.execute();
    }
}

总结

算子之间的关系：
- 一对一
- 重分区
算子串在一起的条件：
- 一对一
- 并行度相同
关于算子链的api:
- 全局禁用算子链：env.disableChaining()
- 某个算子不参与链化：算子A.disableChaining(), 算子A不会与前面和后面的算子串在一起
- 从某个算子开启新链条：算子A.startNewChain(), 算子A不会与前面的算子串在一起, 从A开始正常链化

4. 任务槽(Task Slots)

4.1 任务槽(Task Slots)说明

Flink中每一个TaskManager都是一个JVM进程，它可以启动多个独立的线程，来并行执行多个子任务(subtask)。
很显然，TaskManager的计算资源是有限的，并行的任务越多，每个线程的资源就会越少。那一个TaskManager到底能并行处理多少个任务呢？为了控制并发量，我们需要在TaskManager上对每个任务运行所占用的资源做出明确的划分，这就是所谓的任务槽(task slots)。
每个任务槽(task slot)其实表示了TaskManager拥有计算资源的一个固定大小的子集。这些资源就是用来独立执行一个子任务的。
假如一个TaskManager有三个slot，那么它会将管理的内存平均分成三份，每个slot独自占据一份。这样一来，我们在slot上执行一个子任务时，相当于划定了一块内存"专款专用"，就不需要跟来自其他作业的任务去竞争内存资源了。所以现在我们只要2个TaskManager，就可以并行处理分配好的5个任务了。
Alt text

4.2 任务槽数量的设置

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

yml

taskmanager.numberOfTaskSlots: 1

需要注意的是，slot目前仅仅用来隔离内存，不会涉及CPU的隔离。在具体应用时，可以将slot数量配置为机器的CPU核心数，尽量避免不同任务之间对CPU的竞争。这也是开发环境默认并行度设为机器CPU数量的原因。

4.3 任务对任务槽的共享

在同一个作业中，不同任务节点的并行子任务，就可以放到同一个slot上执行。 Alt text 默认情况下，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.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：并行度为1的word count:
如果我们没有任何并行度设置，而配置文件中默认parallelism.default=1，那么程序运行的默认并行度为1，总共有3个任务。由于不同算子的任务可以共享任务槽，所以最终占用的slot只有1个。9个slot只用了1个，有8个空闲。
示例2：并行度为2的word count:

设置任务并行度的方式有以下三种：

flink-conf.yaml：parallelism.default: 2
flink客户端：./bin/flink run –p 2
执行环境: env.setParallelism(2)
作业并行度设置为2，那么总共有6个任务，共享任务槽之后会占用2个slot。同样，就有7个slot空闲，计算资源没有充分利用。所以可以看到，设置合适的并行度才能提高效率。

示例3：并行度为9的word count

设置任务并行度的方式有以下三种：

flink-conf.yaml：parallelism.default: 9
flink客户端：./bin/flink run –p 9
执行环境：env.setParallelism(9)
怎样设置并行度效率最高呢？当然是需要把所有的slot都利用起来。考虑到slot共享，我们可以直接把并行度设置为9，这样所有27个任务就会完全占用9个slot。这是当前集群资源下能执行的最大并行度，计算资源得到了充分的利用。

示例4：全局并行度为9，并单独设置Sink并行度为1
算子的并行度可以在程序中分别设定：
couts.writeAsCsv(outputpath,"\n"," ").setParallelism(1)
另外再考虑对于某个算子单独设置并行度的场景。例如，如果我们考虑到输出可能是写入文件，那会希望不要并行写入多个文件，就需要设置sink算子的并行度为1。这时其他的算子并行度依然为9，所以总共会有19个子任务。根据slot共享的原则，它们最终还是会占用全部的9个slot，而sink任务只在其中一个slot上执行。

提示

slot特点：

均分隔离内存，不隔离cpu
可以共享：同一个job中，不同算子的子任务，才可以共享; 同一个slot, 同时在运行的前提是属于同一个slot共享组，默认都是"default"

slot数量与并行度的关系

slot是一种静态的概念，表示最大的并发上限。并行度是一种动态概念，表示实际运行占用了几个
要求： slot数量 >= job并行度(算子最大并行度)，job才能运行
值得注意的是如果是yarn模式，需要动态申请资源：申请的TM数量 = job并行度/每个TM的slot数，向上取整。比如session模式中，一开始是0个taskmanager, 0个slot --> 提交一个job后，并行度为10，10/3向上取整，申请4个tm。

5. 作业提交流程

5.1 Standalone会话模式作业提交流程

Alt text 整个作业涉及以下数据流图(物理图看不到，在flink内部)：
逻辑流图(StreamGraph)→ 作业图(JobGraph)→ 执行图(ExecutionGraph)→ 物理图(Physical Graph)

5.2 逻辑流图(StreamGraph)

这是根据用户通过DataStream API编写的代码生成的最初的DAG图，用来表示程序的拓扑结构。这一步一般在客户端完成。

5.3 作业图(JobGraph)

StreamGraph经过优化后生成的就是作业图(JobGraph)，这是提交给JobManager 的数据结构，确定了当前作业中所有任务的划分。主要的优化为：将多个符合条件的节点链接在一起合并成一个任务节点，形成算子链，这样可以减少数据交换的消耗。JobGraph一般也是在客户端生成的，在作业提交时传递给JobMaster。
我们提交作业之后，打开Flink自带的Web UI，点击作业就能看到对应的作业图。

5.4 执行图(ExecutionGraph)

JobMaster收到JobGraph后，会根据它来生成执行图(ExecutionGraph)。ExecutionGraph是JobGraph的并行化版本，是调度层最核心的数据结构。与JobGraph最大的区别就是按照并行度对并行子任务进行了拆分，并明确了任务间数据传输的方式。

5.5 物理图(Physical Graph)

JobMaster生成执行图后，会将它分发给TaskManager；各个TaskManager会根据执行图部署任务，最终的物理执行过程也会形成一张"图"，一般就叫作物理图(Physical Graph)。这只是具体执行层面的图，并不是一个具体的数据结构。
物理图主要就是在执行图的基础上，进一步确定数据存放的位置和收发的具体方式。有了物理图，TaskManager就可以对传递来的数据进行处理计算了。

5.6 Yarn应用模式作业提交流程

Alt text

用户提交Flink作业到Yarn, 首先Resource Manager会启动Application Master(AM), AM在其中一台NodeManager中创建容器，启动Flink。
Flink中的Actor接下来会启动分发器和资源管理器，分发器接收到Flink作业请求会启动JobMaster。
JobMaster会根据作业生成逻辑流图、作业流图、执行流图，并向自己的资源管理器注册和请求slot。
Flink资源管理器会翻译资源请求内容，向Yarn的资源管理器申请资源。
Yarn的资源管理器会根据请求去在其他NodeManager中创建容器启动Flink的TaskManager。
TaskManager启动后会向Flink的资源管理器注册slot。
Flink资源管理器然后再根据请求的slot去分配slot给JobMaster。
TaskManager会找JobMaster, 提供slot基本信息。
JobMaster分配任务，并生成具体执行的物理流图。
TaskManager按照物理流图执行，处理数据流。

Flink运行时架构 ​

1. Standalone会话模式系统架构 ​

2. 并行度(Parallelism) ​

2.1 并行子任务和并行度 ​

2.2 并行度的设置 ​

3. 算子链(Operator Chain) ​

3.1 算子间的数据传输 ​

3.2 一对一(One-to-one，forwarding) ​

3.3 重分区(Redistributing) ​

3.4 合并算子链 ​

4. 任务槽(Task Slots) ​

4.1 任务槽(Task Slots)说明 ​

4.2 任务槽数量的设置 ​

4.3 任务对任务槽的共享 ​

4.4 任务槽和并行度的关系 ​

5. 作业提交流程 ​

5.1 Standalone会话模式作业提交流程 ​

5.2 逻辑流图(StreamGraph) ​

5.3 作业图(JobGraph) ​

5.4 执行图(ExecutionGraph) ​

5.5 物理图(Physical Graph) ​

5.6 Yarn应用模式作业提交流程 ​

Flink运行时架构

1. Standalone会话模式系统架构

2. 并行度(Parallelism)

2.1 并行子任务和并行度

2.2 并行度的设置

3. 算子链(Operator Chain)

3.1 算子间的数据传输

3.2 一对一(One-to-one，forwarding)

3.3 重分区(Redistributing)

3.4 合并算子链

4. 任务槽(Task Slots)

4.1 任务槽(Task Slots)说明

4.2 任务槽数量的设置

4.3 任务对任务槽的共享

4.4 任务槽和并行度的关系

5. 作业提交流程

5.1 Standalone会话模式作业提交流程

5.2 逻辑流图(StreamGraph)

5.3 作业图(JobGraph)

5.4 执行图(ExecutionGraph)

5.5 物理图(Physical Graph)

5.6 Yarn应用模式作业提交流程