3、Flink 部署
Flink 是一个非常灵活的处理框架,它支持多种不同的部署场景,还可以和不同的资源管
理平台方便地集成。
3.1、快速启动一个 Flink 集群
3.1.1、环境配置
Flink 是一个分布式的流处理框架,所以实际应用一般都需要搭建集群环境。我们在进行Flink 安装部署时,需要准备 3 台 Linux 机器,本案例搭建的环境如下:
- 系统环境为 CentOS 7.5 版本。
- 安装 Java 8。
- 安装 Hadoop 集群,Hadoop版本为3.1.3,配置集群节点服务器间时间同步以及免密登录,关闭防火墙。
- 节点服务器 1,IP 地址为 192.168.10.102,主机名为 hadoop102。
- 节点服务器 2,IP 地址为 192.168.10.103,主机名为 hadoop103。
- 节点服务器 3,IP 地址为 192.168.10.104,主机名为 hadoop104。
3.1.2、本地启动
最简单的启动方式,其实是不搭建集群,直接本地启动。本地部署非常简单,直接解压安装包就可以使用,不用进行任何配置, 测试学习使用 。
- 下载安装包
进入 Flink 官网,下载 1.13.0 版本安装包 flink-1.13.0-bin-scala_2.12.tgz,注意此处选用对应 scala 版本为 scala 2.12 的安装包。
- 解压
在 hadoop102节点服务器上创建安装目录/opt/model,将 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 hadoop102.
Starting taskexecutor daemon on host hadoop102.
$ jps
10369 StandaloneSessionClusterEntrypoint
10680 TaskManagerRunner
10717 Jps
- 访问 Web UI
启动成功后,访问 http://hadoop102:8081 ,可以对 flink 集群和任务进行监控管理,如下图所示。
- 关闭集群
如果想要让 Flink 集群停止运行,可以执行以下命令:
$ bin/stop-cluster.sh
Stopping taskexecutor daemon (pid: 10680) on host hadoop102.
Stopping standalonesession daemon (pid: 10369) on host hadoop102.
3.1.3、集群启动
Flink 本地启动非常简单,直接执行 start-cluster.sh 就可以了。如果我们想要扩展成集群,其实启动命令是不变的,主要是需要指定节点之间的主从关系。
Flink 是典型的 Master-Slave 架构的分布式数据处理框架,其中 Master 角色对应着JobManager,Slave 角色则对应 TaskManager。我们对三台节点服务器的角色分配如下图所示。
3.1.3.1、下载并解压安装包
具体操作与上节相同。
3.1.3.2、修改集群配置
- 进入 conf 目录下,修改 flink-conf.yaml 文件,修改 jobmanager.rpc.address 参数为hadoop102,如下所示:
$ cd conf/
$ vim flink-conf.yaml
# JobManager 节点地址.
jobmanager.rpc.address: hadoop102
这就指定了 hadoop102 节点服务器为 JobManager 节点。
- 修改 workers 文件,将另外两台节点服务器添加为本 Flink 集群的 TaskManager 节点,具体修改如下:
$ vim workers
hadoop103
hadoop104
这样就指定了 hadoop103 和 hadoop104 为 TaskManager 节点。
3.1.3.3、 分发安装目录
配置修改完毕后,将 Flink 安装目录发给另外两个节点服务器。
[song@hadoop102 model]$ xsync flink-1.13.0/
3.1.3.4、 启动集群
- 在 hadoop102 节点服务器上执行 start-cluster.sh 启动 Flink 集群:
$ bin/start-cluster.sh
Starting cluster.
Starting standalonesession daemon on host hadoop102.
Starting taskexecutor daemon on host hadoop103.
Starting taskexecutor daemon on host hadoop104.
- 查看进程情况:
[song@hadoop102 flink-1.13.0]$ jps
13859 Jps
13782 StandaloneSessionClusterEntrypoint
[song@hadoop103 flink-1.13.0]$ jps
12215 Jps
12124 TaskManagerRunner
[song@hadoop104 flink-1.13.0]$ jps
11602 TaskManagerRunner
11694 Jps
3.1.3.5、访问 Web UI
启动成功后,同样可以访问 http://hadoop102:8081 对 flink 集群和任务进行监控管理,如下图所示。
这里可以明显看到,当前集群的 TaskManager 数量为 2;由于默认每个 TaskManager 的 Slot数量为 1,所以总 Slot 数和可用 Slot 数都为 2。
3.1.4、向集群提交作业
- 添加打包pom
<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>
-
打包
因为集群中已经具备任务运行所需的所有依赖,所以建议使用 flink_demo-1.0-SNAPSHOT.jar。 -
在 Web UI 上提交作业
- 任务打包完成后,我们打开 Flink 的 WEB UI 页面,在右侧导航栏点击“Submit New Job”,然后点击按钮“+ Add New”,选择要上传运行的 JAR 包,如下图所示
- 点击该 JAR 包,出现任务配置页面,进行相应配置。
主要配置程序入口主类的全类名,任务运行的并行度,任务运行所需的配置参数和保存点路径等,如下图所示,配置完成后,即可点击按钮“Submit”,将任务提交到集群运行。
3.1.5、命令行向集群提交作业
除了通过 WEB UI 界面提交任务之外,也可以直接通过命令行来提交任务。先把 jar 包直接上传到目录 flink-1.13.0 下
- 首先需要启动集群。
$ bin/start-cluster.sh
- 在 hadoop102 中执行以下命令启动 netcat。
$ nc -lk 7777
- 进入到 Flink 的安装路径下,在命令行使用 flink run 命令提交作业。
$ bin/flink run -m hadoop102:8081 -c com.song.wc.StreamWordCount ./flink_demo-1.0-SNAPSHOT.jar
这里的参数 –m 指定了提交到的 JobManager,-c 指定了入口类
- 在浏览器中打开 Web UI,http://hadoop102:8081 查看应用执行情况,如下图所示。
用 netcat 输入数据,可以在 TaskManager 的标准输出(Stdout)看到对应的统计结果。
- 在 log 日志中,也可以查看执行结果,需要找到执行该数据任务的 TaskManager 节点查看日志。
3.2、部署模式
在一些应用场景中,对于集群资源分配和占用的方式,可能会有特定的需求。Flink 为各种场景提供了不同的部署模式,主要有以下三种:
- 会话模式(Session Mode)
- 单作业模式(Per-Job Mode)
- 应用模式(Application Mode)
它们的区别主要在于:集群的生命周期以及资源的分配方式以及应用的 main 方法到底在哪里执行——客户端(Client)还是 JobManager。
3.2.1、会话模式(Session Mode)
会话模式其实最符合常规思维。我们需要先启动一个集群,保持一个会话,在这个会话中通过客户端提交作业,如下图所示。集群启动时所有资源就都已经确定,所以所有提交的作业会竞争集群中的资源。
这样的好处很明显,我们只需要一个集群,就像一个大箱子,所有的作业提交之后都塞进去;集群的生命周期是超越于作业之上的,铁打的营盘流水的兵,作业结束了就释放资源,集群依然正常运行。
当然缺点也是显而易见的:因为资源是共享的,所以资源不够了,提交新的作业就会失败。另外,同一个 TaskManager 上可能运行了很多作业,如果其中一个发生故障导致 TaskManager 宕机,那么所有作业都会受到影响。
会话模式比较适合于单个规模小、执行时间短的大量作业。
3.2.2、单作业模式(Per-Job Mode)
会话模式因为资源共享会导致很多问题,所以为了更好地隔离资源,我们可以考虑为每个提交的作业启动一个集群,这就是所谓的单作业(Per-Job)模式,如下图所示。
单作业模式也很好理解,就是严格的一对一,集群只为这个作业而生。同样由客户端运行应用程序,然后启动集群,作业被提交给 JobManager,进而分发给 TaskManager 执行。作业作业完成后,集群就会关闭,所有资源也会释放。这样一来,每个作业都有它自己的 JobManager管理,占用独享的资源,即使发生故障,它的 TaskManager 宕机也不会影响其他作业。
这些特性使得单作业模式在生产环境运行更加稳定,所以是实际应用的首选模式。 需要注意的是,Flink 本身无法直接这样运行,所以单作业模式一般需要借助一些资源管理框架来启动集群,比如 YARN、Kubernetes。
3.2.3、应用模式(Application Mode)
前面提到的两种模式下,应用代码都是在客户端上执行,然后由客户端提交给 JobManager的。但是这种方式客户端需要占用大量网络带宽,去下载依赖和把二进制数据发送给JobManager;加上很多情况下我们提交作业用的是同一个客户端,就会加重客户端所在节点的资源消耗。
所以解决办法就是,我们不要客户端了,直接把应用提交到 JobManger 上运行。而这也就代表着,我们需要为每一个提交的应用单独启动一个 JobManager,也就是创建一个集群。
这个 JobManager 只为执行这一个应用而存在,执行结束之后 JobManager 也就关闭了,这就是所谓的应用模式,如下图所示。
应用模式与单作业模式,都是提交作业之后才创建集群;
- 单作业模式是通过客户端来提交的,客户端解析出的每一个作业对应一个集群;
- 应用模式下,是直接由 JobManager 执行应用程序的,并且即使应用包含了多个作业,也只创建一个集群。
总结:
- 在会话模式下,集群的生命周期独立于集群上运行的任何作业的生命周期,并且提交的所有作业共享资源。
- 单作业模式为每个提交的作业创建一个集群,带来了更好的资源隔离,这时集群的生命周期与作业的生命周期绑定。
- 应用模式为每个应用程序创建一个会话集群,在 JobManager 上直接调用应用程序的 main()方法。
我们所讲到的部署模式,相对是比较抽象的概念,实际应用时,一般需要和资源管理平台结合起来,选择特定的模式来分配资源、部署应用。接下来,我们就针对不同的资源提供者(Resource Provider)的场景,具体介绍 Flink 的部署方式。
3.3、不同的资源提供者(Resource Provider)的场景下,Flink 的部署方式。
3.3.1、独立模式(Standalone)
独立模式(Standalone)是部署 Flink 最基本也是最简单的方式:所需要的所有 Flink 组件,都只是操作系统上运行的一个 JVM 进程。
独立模式是独立运行的,不依赖任何外部的资源管理平台;当然独立也是有代价的:如果资源不足,或者出现故障,没有自动扩展或重分配资源的保证,必须手动处理。所以独立模式一般只用在开发测试或作业非常少的场景下。
另外,我们也可以将独立模式的集群放在容器中运行。Flink 提供了独立模式的容器化部署方式,可以在 Docker 或者 Kubernetes 上进行部署。
3.3.1.1、会话模式部署
独立模式的特点是不依赖外部资源管理平台,而会话模式的特点是先启动集群、后提交作业。上面启动的WordCount案例就是会话模式部署的。
3.3.1.2、单作业模式部署
Flink 本身无法直接以单作业方式启动集群,一般需要借助一些资源管理平台。所以 Flink 的独立(Standalone)集群并不支持单作业模式部署。
3.3.1.3、应用模式部署
应用模式下不会提前创建集群,所以不能调用 start-cluster.sh 脚本。我们可以使用同样在bin 目录下的 standalone-job.sh 来创建一个 JobManager。
具体步骤如下:
- 进入到 Flink 的安装路径下,将应用程序的 jar 包放到 lib/目录下。
$ cp ./flink_demo-1.0-SNAPSHOT.jar lib/
- 执行以下命令,启动 JobManager。
$ ./bin/standalone-job.sh start --job-classname com.song.wc.StreamWordCount
这里我们直接指定作业入口类,脚本会到 lib 目录扫描所有的 jar 包。
- 同样是使用 bin 目录下的脚本,启动 TaskManager。
$ ./bin/taskmanager.sh start
- 如果希望停掉集群,同样可以使用脚本,命令如下。
$ ./bin/standalone-job.sh stop
$ ./bin/taskmanager.sh stop
3.3.1.4、高可用(High Availability )
分布式除了提供高吞吐,另一大好处就是有更好的容错性。
对于 Flink 而言,因为一般会有多个 TaskManager,即使运行时出现故障,也不需要将全部节点重启,只要尝试重启故障节点就可以了。但是我们发现,针对一个作业而言,管理它的 JobManager 却只有一个,这同样有可能出现单点故障。为了实现更好的可用性,我们需要 JobManager 做一些主备冗余,这就是所谓的高可用(High Availability,简称 HA)。
我们可以通过配置,让集群在任何时候都有一个主 JobManager 和多个备用 JobManagers,如下图所示,这样主节点故障时就由备用节点来接管集群,接管后作业就可以继续正常运行。主备jobManager 实例之间没有明显的区别,每个 JobManager 都可以充当主节点或者备节点
具体配置如下:
- 进入 Flink 的安装路径下的 conf 目录下,修改配置文件: flink-conf.yaml,增加如下配置。
high-availability: zookeeper
high-availability.storageDir: hdfs://hadoop102:9820/flink/standalone/ha
high-availability.zookeeper.quorum: hadoop102:2181,hadoop103:2181,hadoop104:2181
high-availability.zookeeper.path.root: /flink-standalone
high-availability.cluster-id: /cluster_song
- 修改配置文件: masters,配置备用 JobManager 列表。
hadoop102:8081
hadoop103:8081
- 分发修改后的配置文件到其他节点服务器。
- 在/etc/profile.d/my_env.sh 中配置环境变量,export HADOOP_CLASSPATH=
hadoop classpath
注意:
-
需要提前保证 HAOOP_HOME 环境变量配置成功
-
分发到其他节点
具体部署方法如下:- 首先启动 HDFS 集群和 Zookeeper 集群。
- 执行以下命令,启动 standalone HA 集群。
$ bin/start-cluster.sh
- 可以分别访问两个备用 JobManager 的 Web UI 页面。
http://hadoop102:8081 http://hadoop103:8081
- 在 zkCli.sh 中查看谁是 leader。
[zk: localhost:2181(CONNECTED) 1] get /flink- standalone/cluster_atguigu/leader/rest_server_lock
杀死 hadoop102 上的 Jobmanager, 再看 leader。
[zk: localhost:2181(CONNECTED) 7] get /flink-standalone/cluster_atguigu/leader/rest_server_lock
注意: 不管是不是 leader,从 WEB UI 上是看不到区别的, 都可以提交应用。
3.3.2、YARN 模式
独立(Standalone)模式由 Flink 自身提供资源,无需其他框架,这种方式降低了和其他第三方资源框架的耦合性,独立性非常强。
但是,Flink 是大数据计算框架,不是资源调度框架,这并不是它的强项;所以还是应该让专业的框架做专业的事,和其他资源调度框架集成更靠谱。而在目前大数据生态中,国内应用最为广泛的资源管理平台就是 YARN了。
整体来说,YARN 上部署的过程是:
- 客户端把 Flink 应用提交给 Yarn 的 ResourceManager,
- Yarn 的 ResourceManager 会向 Yarn 的 NodeManager 申请容器。
- 在这些容器上,Flink 会部署JobManager 和 TaskManager 的实例,从而启动集群。
- Flink 会根据运行在 JobManger 上的作业所需要的 Slot 数量动态分配 TaskManager 资源。
3.3.2.1、相关准备和配置
在 Flink1.8.0 之前的版本,想要以 YARN 模式部署 Flink 任务时,需要 Flink 是有 Hadoop支持的。从 Flink 1.8 版本开始,不再提供基于 Hadoop 编译的安装包,若需要 Hadoop 的环境支持,需要自行在官网下载 Hadoop 相关版本的组件 flink-shaded-hadoop-2-uber-2.7.5-10.0.jar,并将该组件上传至 Flink 的 lib 目录下。
在 Flink 1.11.0 版本之后,增加了很多重要新特性,其中就包括增加了对Hadoop3.0.0以及更高版本Hadoop的支持,不再提供“flink-shaded-hadoop-*”jar 包,而是通过配置环境变量完成与 YARN 集群的对接。
在将 Flink 任务部署至 YARN 集群之前,需要确认集群是否安装有 Hadoop,保证 Hadoop版本至少在 2.2 以上,并且集群中安装有 HDFS 服务。
具体配置步骤如下:
- 按照之前的描述,下载并解压安装包,并将解压后的安装包重命名为 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_CLASSPATH。
- 启动 Hadoop 集群,包括 HDFS 和 YARN。
[song@hadoop102 ~]$ start-dfs.sh
[song@hadoop103 ~]$ start-yarn.sh
分别在 3 台节点服务器查看进程启动情况。
[song@hadoop102 ~]$ jps
5190 Jps
5062 NodeManager
4408 NameNode
4589 DataNode
[song@hadoop103 ~]$ jps
5425 Jps
4680 ResourceManager
5241 NodeManager
4447 DataNode
[song@hadoop104 ~]$ jps
4731 NodeManager
4333 DataNode
4861 Jps
4478 SecondaryNameNode
- 进入 conf 目录,修改 flink-conf.yaml 文件,根据需要修改以下配置,若在提交命令中不特定指明,这些配置将作为默认配置。
$ cd /opt/module/flink-1.13.0-yarn/conf/
$ vim flink-conf.yaml
jobmanager.memory.process.size: 1600m
taskmanager.memory.process.size: 1728m
taskmanager.numberOfTaskSlots: 8
parallelism.default: 1
3.3.2.2、会话模式部署
YARN 的会话模式与独立集群略有不同,需要首先申请一个 YARN 会话(YARN session)来启动 Flink 集群。具体步骤如下
3.3.2.2.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 或者命令行两种方式提交作业。
2021-06-03 15:54:27,069 INFO org.apache.flink.yarn.YarnClusterDescriptor [] - YARN application has been deployed successfully.
2021-06-03 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
3.3.2.2.2、提交作业
-
通过 Web UI 提交作业
这种方式比较简单,与上文所述 Standalone 部署模式基本相同。 -
通过命令行提交作业
- 将 Standalone 模式讲解中打包好的任务运行 JAR 包上传至集群
- 执行以下命令将该任务提交到已经开启的 Yarn-Session 中运行。
$ bin/flink run -c com.song.wc.StreamWordCount flink_demo-1.0-SNAPSHOT.jar
客户端可以自行确定 JobManager 的地址,也可以通过-m 或者-jobmanager 参数指定JobManager 的地址,JobManager 的地址在 YARN Session 的启动页面中可以找到。
-
任务提交成功后,可在 YARN 的 Web UI 界面查看运行情况。
如上图所示,从图中可以看到我们创建的 Yarn-Session 实际上是一个 Yarn 的Application,并且有唯一的 Application ID。 -
也可以通过 Flink 的 Web UI 页面查看提交任务的运行情况,如下图所示
3.3.2.3、单作业模式部署
在 YARN 环境中,由于有了外部平台做资源调度,所以我们也可以直接向 YARN 提交一个单独的作业,从而启动一个 Flink 集群。
- 执行命令提交作业。
$ bin/flink run -d -t yarn-per-job -c com.song.wc.StreamWordCount flink_demo-1.0-SNAPSHOT.jar
早期版本也有另一种写法:
$ bin/flink run -m yarn-cluster -c com.song.wc.StreamWordCount flink_demo-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 集群也会停掉。
3.3.2.4、应用模式部署
应用模式同样非常简单,与单作业模式类似,直接执行 flink run-application 命令即可。
- 执行命令提交作业。
$ bin/flink run-application -t yarn-application -c om.song.wc.StreamWordCount flink_demo-1.0-SNAPSHOT.jar
- 在命令行中查看或取消作业。
$ ./bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY
$ ./bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY <jobId>
- 也可以通过 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,而不需要单独发送到集群,这就使得作业提交更加轻量了。
3.3.2.4、高可用模式部署
YARN 模式的高可用和独立模式(Standalone)的高可用原理不一样。Standalone 模式中, 同时启动多个 JobManager, 一个为“领导者”(leader),其他为“后备”(standby), 当 leader 挂了, 其他的才会有一个成为 leader。
而 YARN 的高可用是只启动一个 Jobmanager, 当这个 Jobmanager 挂了之后, YARN 会再次启动一个, 所以其实是利用的 YARN 的重试次数来实现的高可用。
- 在 yarn-site.xml 中配置。
<property>
<name>yarn.resourcemanager.am.max-attempts</name>
<value>4</value>
<description>The maximum number of application master execution attempts.</description>
</property>
注意: 配置完不要忘记分发, 和重启 YARN。
- 在 flink-conf.yaml 中配置。
yarn.application-attempts: 3
high-availability: zookeeper
high-availability.storageDir: hdfs://hadoop102:9820/flink/yarn/ha
high-availability.zookeeper.quorum: hadoop102:2181,hadoop103:2181,hadoop104:2181
high-availability.zookeeper.path.root: /flink-yarn
- 启动 yarn-session
- 杀死 JobManager, 查看复活情况。
注意: yarn-site.xml 中配置的是 JobManager 重启次数的上限, flink-conf.xml 中的次数应该小于这个值。