Flink5：Flink运行架构（Slot和并行度）

Posted 2023-04-17

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Flink5：Flink运行架构（Slot和并行度）相关的知识，希望对你有一定的参考价值。

参考技术A 首先来说slots，在之前的学习，我们知道了每一个TaskManager都是干活的进程，它其实就是一JVM进程，那么在干活的过程中，一个JVM进程还能做多线程处理，所以它的内部是：在不同的线程上去执行一个或者多个它的子任务。而这批子任务，这个线程到底能执行多少个呢，为了控制内部线程运行子任务的个数，就提出了slots概念。slots就是TaskManager的固定大小资源的一个集合。ResourceManager在做资源分配管理的时候，最小的单位就是slot。那么接下来就看下ResourceManager是如何分配slot的。
一般的方式，我们按照每一个TaskManager机器的性能，它所含有的资源来配置slot。slot相当于它所有资源的一个子集，这个子集在执行过程中，就是一个隔离开的独立的子任务（线程）。相当于是用slot把不同的子任务之间做了一个隔离。如果机器的内存很大，cpu数量也多，那么就可以让它同时并行执行任务分配更多的slot。
第一点我们要注意的是，slot隔离主要是对内存的隔离，CPU本身是不做隔离的，CPU在不同的slot之间是可以共享的。slot内存是平均分配的，比如机器上有16G内存，如果划分4个slot的话，那每个slot就是4G内存了。如果每个slot内存太小的话，任务就执行不下去了，内存直接被撑爆这个是完全有可能的。
所以在划分内存方面需要我们去考量的，我们要根据我们执行任务的复杂程度，占用资源的角度和我们本身的机器它本身所有的资源大小做一个整体的分配。确定TaskManager到底分成几个slot。现在我们大部分机器其实内存还好，为了避免不同的slot之间共享CPU导致我们资源本身负载的程度不高，这时我们往往按照CPU的数量来分配多少个slot。
第二点要注意的是，默认情况下，Flink允许子任务共享slot，即使它们是不同人物的子任务。这样的结果是，一个slot可以保存作业的整个管道。Task Slot是静态的概念，是指TaskManager具有并发执行能力。
slot分配目的：我们其实是想要资源能够更平均地分配到我们的任务上，能够让我们的资源利用率能够更高。
举例说明，比如在我们所有的算子操作里面，有一些Source和map这样的任务，它相对比较简单，我们把它们叫做非资源密集型的，对资源占用率不是那么高，另外一些任务，比如keyBy()/window()这些聚合操作相对来讲，资源占用率较高，它是资源密集型任务。所以为了防止非资源密集型任务一直处在空闲的状态（等待状态），为了让这些空闲的资源能够利用起来，我们可以将所有的子任务都分配到各个slot上，这样每个slot里有资源密集型任务和非资源密集型任务。都会有简单和复杂的任务，这样所有的slot资源利用率都差不多，就都不会空闲了。
允许slot共享的好处：
1.可以获得更好的资源利用率，不会出现slot之间闲的闲，累的累这种情况。可以让他们每一个slot都差不多忙，具体每个slot里面先忙什么后忙什么那就看自己调度看执行顺序了。
2.JobManager拿到任务执行计划后，它如何确定到底需要多少个slot，这时它只要看整个作业里面，并行度最高的那个算子设置的并行度就可以了，只要满足它的需求，别的就都能满足了。
下面图片展示了TaskManager分配slot任务和优化后的对比情况：
优化前：

优化后：

Flink流式计算从入门到实战二

文章目录

三、Flink运行架构

Flink流式计算实战专题二

==楼兰

三、Flink运行架构

这一章重点是分析清楚运行架构以及并行度与slot的分配

1、JobManager和TaskManager

从之前的环境搭建过程中，也能够看到， Flink中的节点可以分为JobManager和TaskManager。

JobManager处理器也称为Master，用于协调分布式任务执行。他们用来调度task进行具体的任务。TaskManager处理器也称为Worker，用于实际执行任务。

一个有效的Flink集群中可以包含多个JobManager组成高可用集群，也可以有多个TaskManager进行并行计算。他们可以直接在物理机上启动，也可以通过像Yarn这样的资源调度框架启动。

每一个处理器都是一个单独的JVM进程，也可以通过配置的方式管理他们占用的内存资源。在flink-conf.yaml配置文件中，可以通过jobmanager.memory.process.size属性配置jobmanager占用的内存大小，taskmanager.memory.process.size属性配置每个taskmanager占用的内存大小。这个内存大小包含了JVM占用的堆内存以及堆外的元数据区和堆外直接内存的大小。这些参数也可以在提交任务的时候进行干预。

在一个典型的yarn集群中，JobManager在接收到任务时，整体执行的流程会是这样。

客户端会往JobManager提交任务，JobManager会往ResouceManager申请资源，当资源足够时，再将任务分配给集群中的TaskManager去执行。

只不过在Standalone模式下，这个ResourceManager是由Flink自己担任的。而在Yarn模式下，则是转为由Yarn来担任ResourceManager角色。

jobManager的内存架构：

Off-Heap Memory是Flink自行管理的一块内存，在JVM内存之外的直接内存Direct或者本地内存Native。配置JobManager的内存主要是两个参数 jobmanager.memory.flink.size 配置flink执行应用的内存大小和jobmanager.memory.process.size flink集群运行以及应用执行的总内存大小。

更多详细配置参见官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.12/deployment/memory/mem_setup_jobmanager.html

Taskmanager的内存架构：

通常情况下，对于Taskmanager只需要配置一个总的内存大小即可。 taskmanager.memory.flink.size配置flink执行应用程序的总内存大小， taskmanager.memory.process.size 配置整个flink集群占用的内存大小。这两个属性不建议同时配置。

Flink也支持对taskManager的内存分布进行深度定制。TaskManager的Off-Heap Memory分为Managed Memory和Direct Memory两部分。 Managed Memory是Flink的TaskManager自行管理的一块内存，主要有三个用途：

1、Streaming Job可以用来缓存状态后端。比如RocksDB 状态后端
2、 Batch Jobs主要用来进行排序、分组等结果的中间缓存。
3、User Defined Function可以直接用。官网上说是给Python的processor用的。

Direct Memory部分分为Framework off-Heap 、Task off-Heap 、 Network三个部分。其中

1、Framework Heap Memory : Flink框架执行所需要的JVM堆内存，可以由属性 taskmanager.memory.framework.heap.size 配置。
2、Task off-Heap: 执行Flink程序所需要占用的JVM堆内存，可以由属性taskmanager.memory.task.heap.size 配置。
3、Network: 这一部分内存主要是为任务执行过程中的网络数据交互预留。这一部分是动态分配的，可以由以下三个属性进行定制： taskmanager.memory.network.min、taskmanager.memory.network.max、taskmanager.memory.network.fraction

这里只列出了使用频率相对比较高的几个配置信息。

更多详细配置参见官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/deployment/memory/mem_setup_tm.html

2、并发度与Slots

每一个TaskManager是一个独立的JVM进程，他可以在独立的线程上执行一个或多个任务task。为了控制一个taskManager能接收多少个task，TaskManager上就会划分出多个slot来进行控制。每个slot表示的是TaskManager上拥有资源的一个固定大小的子集。flink-conf.yaml配置文件中的taskmanager.numberOfTaskSlots属性就配置了配个taskManager上有多少个slot。默认值是1，所以我们之前搭建的集群，有3个taskManager，集群内总共就只有3个slot。这些slot之间的内存管理也就是数据是相互隔离的。而这些slot其实都是在同一个JVM进程中，所以这里的隔离并不涉及到CPU等其他资源的隔离。

Task Slot是一个静态的概念，代表的是TaskManager具有的并发执行能力。另外还有一个概念并行度 parallelism就是一个动态的概念，表示的是运行程序时实际需要使用的并发能力。这个是可以在flink程序中进行控制的。如果集群提供的slot资源不够，那程序就无法正常执行下去，会表现为任务阻塞或者超时异常。

程序运行时的parallelism管理有三个地方可以配置，优先级最低的是在flink-conf.yaml文件中的parallelism.default这个属性，默认值是1。优先级较高的是在提交任务时可以指定任务整体的并行度要求。这个并行度可以在提交任务的管理页面和命令行中添加。优先级最高的是在程序中指定的并行度。在flink的应用程序中，几乎每一个分布式操作都可以定制单独的并行度。这到底是是怎么回事呢？那现在我们就开发一个简单的flink应用了解一下。

3、开发环境搭建

flink提供了java和scala两套客户端API，我们这里采用java进行演示。

首先创建一个maven工程，在pom.xml文件中，引入客户端的依赖

		<dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.12.5</version>
        </dependency>
		<dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>1.12.5</version>
        </dependency>
		<dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>1.12.5</version>
        </dependency>

后面这个依赖中最后的2.12表示是对应的scala版本。

然后就可以开发一个简单的flink应用程序。

package com.roy.flink.streaming;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class SocketWordCount {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String host = parameterTool.get("host");
        final int port = parameterTool.getInt("port");

        final DataStreamSource<String> inputDataStream = environment.socketTextStream(host, port);

        final DataStream<Tuple2<String, Integer>> wordcounts = inputDataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                final String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(new Tuple2<String, Integer>(word, 1));
                }
            }
        })
                .setParallelism(2)
                .keyBy(value -> value.f0)
                .sum(1)
                .setParallelism(3);
        wordcounts.print();
        environment.execute("stream word count");

    }
}

这个程序的作用就是连接一个socket服务端，读取socketStream文本流，然后进行最为经典的WordCount操作。

首先，执行这个测试程序需要有一个socket服务端。我们可以在hadoop01机器上使用nc指令模拟一个。 nc -lk 7777 在本地7777端口建立一个socket服务端。

然后，在本地的IDEA运行配置页面，指令要连接的host和port。

配置完成后，就可以在本地直接运行这个示例了。这也就是Flink所谓的LOCAL模式。

这个执行结果就是最终的wordcount结果。但是，这里面有个有趣的现象。对hello的次数统计，是从1，一步一步统计到3的，而不是一次性统计到3。其实这也体现了流式计算的特点。这些词其实是一个一个统计的。

然后要注意一下我们代码中进行了多次setParallelism操作。在这个演示过程中，暂时没有体现出什么作用。在后续的演示中会有用。

4、提交到集群执行

这种本地执行的方式显然不具备生产使用的要求。我们可以使用maven进行编译，将这个代码编译成一个jar包，FlinkDemo-1.0.jar。

然后使用之前搭建的standalone模式的管理控制台来提交任务。

选standalone模式是因为这个模式的资源控制比较简单。

访问控制台，打开 Submit New Job页面，选择 +Add New 按钮，提交jar包。

单独提供一个jar包还并不足以启动任务，因为启动任务还需要指定任务的入口。选择这个FlinkDemo-1.0.jar，继续配置一个任务。

在这里注意下，提交任务时可以指定这个应用整体的Parallism 并行度。

点击提交，就可以开启一个任务。在running job页面就可以看到正在执行的任务 stream word count。选择这个任务，就能看到任务的执行情况。

这个数据流图展示了整个这个应用的具体执行的步骤。这些步骤整体就构成了数据流图。下面的数据流量会统计每个步骤经过的数据流量。在hadoop01机器上的nc服务中敲入字符，这个数据流量与记录数就会不断增加。

最后应用中通过print打印出来的消息会输入到应用的标准输出控制台。控制台的内容可以在TaskManagers菜单中查看。

5、并行度分析

这里我们重点分析每个蓝色方块下面的Parallelism参数。这里列出了每个步骤所占用的slot数量。而这里统计出来的slot数量就是按照之前所说的优先级确定的。整体优先级是这样。

程序中指定 > 提交任务时指定 > flink-conf.yaml中指定

注意下，这里面第一个和第四个步骤，他们的并行度其实是在具体实现时固定的指定了为1。

然后，我们回到Overview页面，查看下整体的slot情况。

接下来可以看到，我们这个job总共需要7个slot，但是集群中只有3个slot，程序也正常执行起来了。这也体现了slot复用的效果。也就是说slot可以在不同的执行步骤中处理不同的任务。只要集群资源能够支撑应用最大的并行度要求，整个应用就可以运行起来。实际上，Flink对于这个数据流图还会有一些自己的优化，例如某些相邻的操作，他们的并行度相同，任务也不是很复杂时，flink会将这些相邻的步骤进行合并。

这些slot在同一个任务内部是可以不断复用的，但是在不同的任务之间，是不能共用的。所以，这时可以看到，集群中仅有的3个slot已经全部被这个stream word count应用给占满了，如果需要再启动应用，就无法执行了。这时jobmanager会不断的尝试重新申请slot，如果集群中有空出来的slot，那就可以分配给应用。如果一直申请不下来，jobmanager会不断重试，默认每重试10次就会休息一点时间，过后再继续申请。如果在attached模式下，在客户端可以很清晰的看到这个过程。

6、Flink整体运行流程

然后我们再回头来看Flink官方提供的集群结构图就比较清晰了。

客户端

对于Flink，可以通过执行一个Java/Scala程序，或者通过./bin/flink run … 指令启动一个客户端。客户端将把sataflow提交给JobManager。客户端的主要作用其实就是构建好一个Dataflow graph或者也称为JobGraph，然后提交给客户端。而这个JobGraph如果在客户端本地构建，这就是Per-job模式，如果是提交到JobManager由Flink集群来构建，这就是Application模式。然后将提交完成后，客户端可以选择立即结束，这就是detached模式。也可以选择继续执行，来不断跟踪JobManager反馈的任务执行情况，这就是默认的attached模式。

JobManager

JobManager会首先接收到客户端提交的应用程序。这个应用程序整体会包含几个部分：作业图JobGraph，数据流图logic dataflow graph以及打包了所有类库以及资源的jar包。这些资源都将分发给所有的TaskManager去真正执行任务。

JobGraph相当于是一个设计图，之前Yarn的Per-job模式，往集群提交的就是这个JobGraph。JobManger会把JobGraph转换成一个物理层面的数据流图，这个图被叫做执行图 ExecutionGraph，这其中包含了所有可以并发执行的任务，相当于是一个执行计划。接下来JobGraph会向资源管理器例如Yarn的ResourceManager 请求执行任务必要的资源，这些资源会表现为TaskManager上的slot插槽。一旦获得了足够多的资源，就会将执行图分发到真正运行任务的TaskManager上。而在运行过程中，JobManager还会负责所有需要中央协调的操作，例如反馈任务执行结果，协调检查点备份，协调故障恢复等。

JobManager整体上由三个功能模块组成：

ResourceManager

ResourceManager在Flink集群中负责申请、提供和注销集群资源，并且管理task slots。Flink中提供了非常多的ResourceManager实现，比如Yarn，Mesos，K8s和standalone模式。在standalone模式下，ResourceManager只负责在TaskManager之间协调slot的分配，而TaskManager的启动只能由TaskManager自己管理。
Dispatcher

Dispatcher模块提供了一系列的REST接口来提交任务，Flink的控制台也是由这个模块来提供。并且对于每一个执行的任务，Dispatcher会启动一个新的JobMaster，来对任务进行协调。
JobMaster

一个JobMaster负责管理一个单独的JobGraph。Flink集群中，同一时间可以运行多个任务，每个任务都由一个对应的JobMaster来管理。

一个集群中最少有一个JobManager。而在高可用部署时，也可以有多个JobManager。这些JobManager会选举出一个作为Leader。而其他的节点就出于StandBy备用的状态。

TaskManager

TaskManager也成为Worker。每个TaskManager上可以有一个或多个Slot。这些Slot就是程序运行的最小单元。在flink.conf.yaml文件中通过taskmanager.numberOfTaskSlots属性进行配置。