七、Flink部署详解

“

Flink提交作业和执行任务，需要几个关键组件：客户端（Client）、作业管理器（JobManager）和任务管理器（TaskManager）。我们的代码由客户端获取并做转换，之后提交给JobManger。所以JobManager就是Flink集群里的“管事人”，对作业进行中央调度管理；而它获取到要执行的作业后，会进一步处理转换，然后分发任务给众多的TaskManager。这里的TaskManager，就是真正“干活的人”，数据的处理操作都是它们来做的。

”

“

Flink是一个非常灵活的处理框架，它支持多种不同的部署场景，还可以和不同的资源管理平台方便地集成。所以接下来我们会先做一个简单的介绍，让大家有一个初步的认识，之后再展开讲述不同情形下的Flink部署。

”

7.1 快速启动一个Flink集群

7.1.1 环境配置

Flink是一个分布式的流处理框架，所以实际应用一般都需要搭建集群环境。我们在进行Flink安装部署的学习时，需要准备3台Linux机器。具体要求如下：

“

系统环境为CentOS 7.5版本。

安装Java 8。

安装Hadoop集群，Hadoop建议选择Hadoop 2.7.5以上版本。

配置集群节点服务器间时间同步以及免密登录，关闭防火墙。

本书中三台服务器的具体设置如下：

节点服务器1，IP地址为192.168.10.100，主机名为hadoop01。

节点服务器2，IP地址为192.168.10.101，主机名为hadoop02。

节点服务器3，IP地址为192.168.10.102，主机名为hadoop103。

”

7.1.2 本地启动

最简单的启动方式，其实是不搭建集群，直接本地启动。本地部署非常简单，直接解压安装包就可以使用，不用进行任何配置；一般用来做一些简单的测试。

具体安装步骤如下：

• 下载安装包

“

进入Flink官网，下载1.13.0版本安装包flink-1.13.0-bin-scala_2.12.tgz，注意此处选用对应scala版本为scala 2.12的安装包。

”

• 解压

在hadoop100节点服务器上创建安装目录/opt/module，将flink安装包放在该目录下，并执行解压命令，解压至当前目录。

$ tar -zxvf flink-1.13.0-bin-scala_2.12.tgz -C /opt/module/
flink-1.13.0/
flink-1.13.0/log/
flink-1.13.0/LICENSE
flink-1.13.0/lib/
……

• 启动

进入解压后的目录，执行启动命令，并查看进程。

$ cd flink-1.13.0/
$ bin/start-cluster.sh 
Starting cluster.
Starting standalonesession daemon on host hadoop100.
Starting taskexecutor daemon on host hadoop100.
$ jps
10589 StandaloneSessionClusterEntrypoint
10697 TaskManagerRunner
10898 Jps

• 访问Web UI

启动成功后，访问http://hadoop100:8081，可以对flink集群和任务进行监控管理，如图所示。

• 关闭集群如果想要让Flink集群停止运行，可以执行以下命令：

$ bin/stop-cluster.sh 
Stopping taskexecutor daemon (pid: 10680) on host hadoop100.
Stopping standalonesession daemon (pid: 10369) on host hadoop100.

7.1.3 集群启动

可以看到，Flink本地启动非常简单，直接执行start-cluster.sh就可以了。如果我们想要扩展成集群，其实启动命令是不变的，主要是需要指定节点之间的主从关系。

Flink是典型的Master-Slave架构的分布式数据处理框架，其中Master角色对应着JobManager，Slave角色则对应TaskManager。我们对三台节点服务器的角色分配如表所示。

集群角色分配

具体安装部署步骤如下：

1. 下载并解压安装包

具体操作与上节相同。

2. 修改集群配置

• 进入conf目录下，修改flink-conf.yaml文件，修改jobmanager.rpc.address参数为hadoop100，如下所示：

$ cd conf/
$ vim flink-conf.yaml
jobmanager.rpc.address: hadoop100

这就指定了hadoop100节点服务器为JobManager节点。

• 修改workers文件，将另外两台节点服务器添加为本Flink集群的TaskManager节点，具体修改如下：

$ vim workers 
hadoop101
hadoop102

这样就指定了hadoop101和hadoop102为TaskManager节点。

• 另外，在flink-conf.yaml文件中还可以对集群中的JobManager和TaskManager组件进行优化配置，主要配置项如下：

jobmanager.memory.process.size：对JobManager进程可使用到的全部内存进行配置，包括JVM元空间和其他开销，默认为1600M，可以根据集群规模进行适当调整。
taskmanager.memory.process.size：对TaskManager进程可使用到的全部内存进行配置，包括JVM元空间和其他开销，默认为1600M，可以根据集群规模进行适当调整。
taskmanager.numberOfTaskSlots：对每个TaskManager能够分配的Slot数量进行配置，默认为1，可根据TaskManager所在的机器能够提供给Flink的CPU数量决定。所谓Slot就是TaskManager中具体运行一个任务所分配的计算资源。
parallelism.default：Flink任务执行的默认并行度，优先级低于代码中进行的并行度配置和任务提交时使用参数指定的并行度数量。
关于Slot和并行度的概念，我们会在下一章做详细讲解。

3. 分发安装目录

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

$ scp -r ./flink-1.13.0 atguigu@hadoop103:/opt/module
$ scp -r ./flink-1.13.0 atguigu@hadoop104:/opt/module

4. 启动集群

• 在hadoop100节点服务器上执行start-cluster.sh启动Flink集群：

$ bin/start-cluster.sh 
Starting cluster.
Starting standalonesession daemon on host hadoop100.
Starting taskexecutor daemon on host hadoop101.
Starting taskexecutor daemon on host hadoop102.

• 查看进程情况：

[hadoop@hadoop100 flink-1.13.0]$ jps
13859 Jps
13782 StandaloneSessionClusterEntrypoint
[hadoop@hadoop101 flink-1.13.0]$ jps
12215 Jps
12124 TaskManagerRunner
[hadoop@hadoop102 flink-1.13.0]$ jps
11602 TaskManagerRunner
11694 Jps

5. 访问Web UI

启动成功后，同样可以访问http://hadoop100:8081对flink集群和任务进行监控管理，如图所示。

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

7.1.4 向集群提交作业

在上一章中，我们已经编写了词频统计的批处理和流处理的示例程序，并在开发环境的模拟集群上做了运行测试。现在既然已经有了真正的集群环境，那接下来我们就要把作业提交上去执行了。

本节我们将以流处理的程序为例，演示如何将任务提交到集群中进行执行。具体步骤如下。

1. 程序打包

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

<build>
   <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-assembly-plugin</artifactId>
        <version>3.0.0</version>
        <configuration>
          <descriptorRefs>
            <descriptorRef>jar-with-dependencies</descriptorRef>
          </descriptorRefs>
        </configuration>
        <executions>
          <execution>
            <id>make-assembly</id>
            <phase>package</phase>
            <goals>
              <goal>single</goal>
            </goals>
          </execution>
        </executions>
      </plugin>
    </plugins>
</build>

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

[INFO] -----------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] -----------------------------------------------------------------------
[INFO] Total time: 21.665 s
[INFO] Finished at: 2021-06-01T17:21:26+08:00
[INFO] Final Memory: 141M/770M
[INFO] -----------------------------------------------------------------------

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

2. 在Web UI上提交作业

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

上传完成后，如图所示：

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

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

• 点击该任务，可以查看任务运行的具体情况，也可以通过点击“Cancel Job”结束任务运行，如图所示。

3. 命令行提交作业

除了通过WEB UI界面提交任务之外，也可以直接通过命令行来提交任务。这里为方便起见，我们可以先把jar包直接上传到目录flink-1.13.0下

• 首先需要启动集群。

• 在hadoop100中执行以下命令启动netcat。

• 进入到Flink的安装路径下，在命令行使用flink run命令提交作业。

$ bin/flink run -m hadoop100:8081 -c com.liuhao.WordCountL ./FlinkTutorial-1.0-SNAPSHOT.jar --host hadoop102 --port 7777

这里的参数 –m指定了提交到的JobManager，-c指定了入口类。

• 在浏览器中打开Web UI，http://hadoop100:8081查看应用执行情况，如图所示。

用netcat输入数据，可以在TaskManager的标准输出（Stdout）看到对应的统计结果。

• 在log日志中，也可以查看执行结果，需要找到执行该数据任务的TaskManager节点查看日志。

$ cat flink-atguigu-taskexecutor-0-hadoop102.out
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/module/flink-1.13.0/lib/log4j-slf4j-impl-2.12.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/module/hadoop-3.1.3/share/hadoop/common/lib/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
(hello,1)
(hello,2)
(flink,1)
(hello,3)
(scala,1)

7.2 部署模式

在一些应用场景中，对于集群资源分配和占用的方式，可能会有特定的需求。Flink为各种场景提供了不同的部署模式，主要有以下三种：

“会话模式（Session Mode）
单作业模式（Per-Job Mode）
应用模式（Application Mode）
”

它们的区别主要在于：集群的生命周期以及资源的分配方式；以及应用的main方法到底在哪里执行——客户端（Client）还是JobManager。

7.2.1 会话模式（Session Mode）

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

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

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

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

单作业模式也很好理解，就是严格的一对一，集群只为这个作业而生。同样由客户端运行应用程序，然后启动集群，作业被提交给JobManager，进而分发给TaskManager执行。作业作业完成后，集群就会关闭，所有资源也会释放。

这些特性使得单作业模式在生产环境运行更加稳定，所以是实际应用的首选模式。

需要注意的是，Flink本身无法直接这样运行，所以单作业模式一般需要借助一些资源管理框架来启动集群，比如YARN、Kubernetes。

7.2.3 应用模式（Application Mode）

前面提到的两种模式下，应用代码都是在客户端上执行，然后由客户端提交给JobManager的。但是这种方式客户端需要占用大量网络带宽，去下载依赖和把二进制数据发送给JobManager；加上很多情况下我们提交作业用的是同一个客户端，就会加重客户端所在节点的资源消耗。

所以解决办法就是，我们不要客户端了，直接把应用提交到JobManger上运行。而这也就代表着，我们需要为每一个提交的应用单独启动一个JobManager，也就是创建一个集群。这个JobManager只为执行这一个应用而存在，执行结束之后JobManager也就关闭了，这就是所谓的应用模式。

应用模式与单作业模式，都是提交作业之后才创建集群；单作业模式是通过客户端来提交的，客户端解析出的每一个作业对应一个集群；而应用模式下，是直接由JobManager执行应用程序的，并且即使应用包含了多个作业，也只创建一个集群。

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

7.3 独立模式（Standalone）

独立模式（Standalone）是部署Flink最基本也是最简单的方式：所需要的所有Flink组件，都只是操作系统上运行的一个JVM进程。

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

7.3.1 会话模式部署

我们在第7.1节用的就是独立（Standalone）集群的会话模式部署。

7.3.2 单作业模式部署

在7.4.2节中我们提到，Flink本身无法直接以单作业方式启动集群，一般需要借助一些资源管理平台。所以Flink的独立（Standalone）集群并不支持单作业模式部署。

7.3.3 应用模式部署

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

具体步骤如下：

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

$ cp ./FlinkTutorial-1.0-SNAPSHOT.jar lib/

• 执行以下命令，启动JobManager。

$ ./bin/standalone-job.sh start --job-classname com.liuhao.WordCountL

这里我们直接指定作业入口类，脚本会到lib目录扫描所有的jar包。

• 同样是使用bin目录下的脚本，启动TaskManager。

$ ./bin/taskmanager.sh start

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

$ ./bin/standalone-job.sh stop
$ ./bin/taskmanager.sh stop

7.4 YARN模式

YARN上部署的过程是：客户端把Flink应用提交给Yarn的ResourceManager, Yarn的ResourceManager会向Yarn的NodeManager申请容器。在这些容器上，Flink会部署JobManager和TaskManager的实例，从而启动集群。Flink会根据运行在JobManger上的作业所需要的Slot数量动态分配TaskManager资源。

7.4.1 相关准备和配置

在将Flink任务部署至YARN集群之前，需要确认集群是否安装有Hadoop，保证Hadoop版本至少在2.2以上，并且集群中安装有HDFS服务。

具体配置步骤如下：

• 按照7.1节所述，下载并解压安装包，并将解压后的安装包重命名为flink-1.13.0-yarn，本节的相关操作都将默认在此安装路径下执行。
• 配置环境变量，增加环境变量配置如下：

$ sudo vim /etc/profile.d/my_env.sh
HADOOP_HOME=/opt/module/hadoop-2.7.5
export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin
export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop
export HADOOP_CLASSPATH=`hadoop classpath`

• 启动Hadoop集群，包括HDFS和YARN。

[hadoop@hadoop100 ~]$ start-dfs.sh
[hadoop@hadoop101 ~]$ start-yarn.sh

• 解压Flink安装包，并进入conf目录，修改flink-conf.yaml文件，修改配置。

7.4.2 会话模式部署

YARN的会话模式与独立集群略有不同，需要首先申请一个YARN会话（YARN session）来启动Flink集群。具体步骤如下：

1. 启动集群

• 启动hadoop集群(HDFS, YARN)。
• 执行脚本命令向YARN集群申请资源，开启一个YARN会话，启动Flink集群。

$ 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或者命令行两种方式提交作业。

2022-11-25 15:54:27,069 INFO  org.apache.flink.yarn.YarnClusterDescriptor          [] - YARN application has been deployed successfully.
2022-11-25 15:54:27,070 INFO  org.apache.flink.yarn.YarnClusterDescriptor          [] - Found Web Interface hadoop104:39735 of application 'application_1622535605178_0003'.
JobManager Web Interface: http://hadoop104:39735

2. 提交作业

• 通过Web UI提交作业

“

这种方式比较简单，与上文所述Standalone部署模式基本相同。

”

• 通过命令行提交作业

“将Standalone模式讲解中打包好的任务运行JAR包上传至集群
执行以下命令将该任务提交到已经开启的Yarn-Session中运行。。
”

$ bin/flink run -c com.liuhao.WordCountL FlinkTutorial-1.0-SNAPSHOT.jar

客户端可以自行确定JobManager的地址，也可以通过-m或者-jobmanager参数指定JobManager的地址，JobManager的地址在YARN Session的启动页面中可以找到。

任务提交成功后，可在YARN的Web UI界面查看运行情况。

从图中可以看到我们创建的Yarn-Session实际上是一个Yarn的Application，并且有唯一的Application ID。

也可以通过Flink的Web UI页面查看提交任务的运行情况。

7.4.3 单作业模式部署

在YARN环境中，由于有了外部平台做资源调度，所以我们也可以直接向YARN提交一个单独的作业，从而启动一个Flink集群

• 执行命令提交作业。

$ bin/flink run -d -t yarn-per-job -c com.liuhao.WordCountL FlinkTutorial-1.0-SNAPSHOT.jar

• 早期版本也有另一种写法：

$ bin/flink run -m yarn-cluster -c com.liuhao.WordCountL FlinkTutorial-1.0-SNAPSHOT.jar

注意这里是通过参数-m yarn-cluster指定向YARN集群提交任务。

• 在YARN的ResourceManager界面查看执行情况。

点击可以打开Flink Web UI页面进行监控：

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

$ ./bin/flink list -t yarn-per-job -Dyarn.application.id=application_XXXX_YY
$ ./bin/flink cancel -t yarn-per-job -Dyarn.application.id=application_XXXX_YY <jobId>

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

7.4.4 应用模式部署

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

• 执行命令提交作业。

$ bin/flink run-application -t yarn-application -c com.liuhao.WordCountL FlinkTutorial-1.0-SNAPSHOT.jar

• 在命令行中查看或取消作业。

$ ./bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY
$ ./bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY

• 也可以通过yarn.provided.lib.dirs配置选项指定位置，将jar上传到远程。

$ ./bin/flink run-application -t yarn-application -Dyarn.provided.lib.dirs="hdfs://myhdfs/my-remote-flink-dist-dir" hdfs://myhdfs/jars/my-application.jar

这种方式下jar可以预先上传到HDFS，而不需要单独发送到集群，这就使得作业提交更加轻量了。

7.5 K8S 模式

容器化部署是如今业界流行的一项技术，基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是Kubernetes（k8s），而Flink也在最近的版本中支持了k8s部署模式。基本原理与YARN是类似的，具体配置可以参见官网说明，这里我们就不做过多讲解了。