Flink 搭建 - Flink On Yarn 集群模式

Apache Flink 1.14 Documentation

2022-04-29 更新，在 Hadoop-2.7.5 版本的 Yarn 环境部署 FLINK-1.13.6

什么是 FLINK ON YARN

FLINK 应用的发布及执行方式非常丰富，本篇博客主要讲述 FLINK ON YARN 这种方式

Apache Hadoop YARN是 Hadoop 的资源管理器，Flink 实现了 YARN 的任务提交及执行接口，所以我们可以把 FLINK 任务发布在 YARN 上，由 YARN 来完成 FLINK 任务在 YARN 上执行期间的资源调度和管理

Flink 可以根据 JobManager 上运行的作业所需的处理槽数动态分配和取消分配 TaskManager 资源

安装部署

① HADOOP 的版本为 2.4.1+

② 下载 FLINK- 1.13.1

③ 配置 HADOOP_CLASSPATH 环境变量

vim /etc/profile
# 增加该内容
export HADOOP_CLASSPATH=`hadoop classpath`
# 刷新环境变量
source /etc/profile

④ 上传解压

cd /opt/server
tar -zxvf flink-1.13.1-bin-scala_2.11.tgz
ln -s /opt/server/flink-1.10.0 /opt/server/flink

⑤ 修改 /yarn-site.xml （3 台都操作）

vim /opt/server/hadoop/etc/hadoop/yarn-site.xml
# 修改以下内容

<!-- 关闭yarn内存检查 -->
<!-- 是否启动一个线程检查每个任务正使用的虚拟内存量，如果任务超出分配值，则直接将其杀掉，默认为 true -->
<!-- 因为对于 flink 使用 yarn 模式下，很容易内存超标，这个时候 yarn 会自动杀掉 job，因此需要关掉-->
<property>
   <name>yarn.nodemanager.pmem-check-enabled</name>
   <value>false</value>
</property>

<property>
   <name>yarn.nodemanager.vmem-check-enabled</name>
   <value>false</value>
</property>

三种部署模式的介绍和区别

For production use, we recommend deploying Flink Applications in the Per-job or Application Mode, as these modes provide a better isolation for the Applications.

注意：在生产中建议使用 Per-job 或 Application Mode 模式部署 Flink 应用程序，这些模式为应用程序提供了更好的隔离. 一个任务启动一个Flink集群, 各个 Flink 集群之间独立运行, 互不影响, 而且每个集群可以单独进行配置

① Application Mode （应用模式）

# 特点
每次递交作业都需要申请一次资源
# 优点
1. main 方法在集群中执行, 不是执行在客户端
2.  运行时所需的依赖项, 并生成 JobGraph 都在Yarn集群中执行, 有效减少客户端带宽、性能的消耗, 从而防止客户端压力过大
3. 允许应用程序包含多个Job, 多个Job顺序执行

② Per-Mode (Job 分离模式)

# 特点
每次递交作业都需要申请一次资源
# 优点
1. 作业运行完成, 资源会立刻被释放, 不会一直占用系统资源
2. 每个作业都是一个Flink集群, 不同资源之间做到了很好的资源隔离, 且不同的作业可以进行单独的配置
# 缺点
1. 每次递交作业都需要申请资源, 会影响执行效率, 因为申请资源需要消耗时间
2. main 方法在客户端执行, 在客户端提交给 JobManager
# 应用场景
适合作业比较少的场景、大作业的场景

② Session Mode（会话模式）

# 特点
所有作业共享集群资源
# 优点
根据指定的资源参数初始化一个Flink集群,资源只需要一次申请
# 缺点
1. 隔离性差, JM 负载瓶颈
2. main 方法在客户端执行
# 应用场景
对延迟非常敏感但运行时长较短的作业

③ Per-Job Mode（Job 分离模式）

启动测试（Application Mode ｜ 应用模式）

该模式将在 YARN 上启动 Flink 集群，其中应用程序 jar 的 main()方法在 YARN 中的 JobManager 上执行， 应用程序完成后，集群将立即关闭

使用 yarn application -kill <ApplicationId> 或取消 Flink 作业来手动停止集群

# 常用的启动命令, 启动后会得到一个 applicatid
./bin/flink run-application \
      -t yarn-application \
      -yjm 1024 \
      -ytm 1024 \
      -d \
      -c org.apache.flink.streaming.examples.socket.SocketWindowWordCount \
      examples/streaming/SocketWindowWordCount.jar \
      --port 9000

# 列出运行在集群中的任务(通过 -Dyarn.application.id 可以查看指定 application.id 中执行的 Job)
./bin/flink list \
      -t yarn-application \
      -Dyarn.application.id=application_XXXX_YY

# 取消job
./bin/flink cancel \
      -t yarn-application \
      -Dyarn.application.id=application_XXXX_YY <jobId>

注意：在应用集群中取消 JOB 会使整个集群停止

为了让整个应用得到更快的响应，我们可以把 jar 等文件上传到指定的 HDFS 路径，这样应用在执行的时候可以直接从指定的路径获取，而不是从 client 上传

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

② Per-Job Mode（Job 分离模式）

启动测试（Session Mode ｜ 会话模式）

会话模式有两种操作模式

# attached mode(默认)
yarn-session.sh 客户端将 Flink 集群提交给 YARN, 但客户端一直在运行, 不会退出 
如果集群失败, 客户端将显示错误; 如果客户端被终止, 它也会发出集群关闭的信号

# detached mode (-d or --detached)
yarn-session.sh 客户端将 Flink 集群提交给 YARN，然后客户端返回并退出 
需要再次调用客户端或 YARN 工具来停止 Flink 集群

注意：会话模式将在 /tmp/.yarn-properties-<username> 中创建一个隐藏的 YARN 属性文件，该文件将在提交作业时被命令行界面用于集群发现。

注意：除了通过 conf/flink-conf.yaml 文件传递配置外，您还可以在提交时使用 -Dkey=value 参数将任何配置传递给 ./bin/yarn-session.sh 客户端

① 启动 Flink 会话（任意一台都操作）

cd /opt/server/flink

# 启动命令
./bin/yarn-session.sh \
  -yjm 1024 \
  -ytm 1024 \
  -d

② 如果我们想重新通过客户端回到会话中，我们可以使用这个命令

# 格式
./bin/yarn-session.sh -id application_XXXX_YY

# 示例
./bin/yarn-session.sh -id application_1646190502615_0097

③ 使用 Flink 提交任务（任意一台都操作）

注意：此时提交的任务都通过该会话（Session）执行，不会再申请 yarn 资源

# 我们在提交 Flink 作业时在命令行界面中手动指定目标 YARN 集群
./bin/flink run \
      -t yarn-session \
      -Dyarn.application.id=application_1646190502615_0098 \
      -d \
      -c org.apache.flink.streaming.examples.socket.SocketWindowWordCount \
      examples/streaming/SocketWindowWordCount.jar \
      --port 9000

注意：提交 Job 也可以通过 WEB-UI 界面提交任务

④ 关闭指定的 Job

⑤ 关闭 Session

yarn application -kill application_XXXX_YY

启动测试（Per-Job Mode ｜Job分离模式）

Per-job Cluster 模式会在 YARN 上启动一个 Flink 集群，然后在本地运行提供的应用程序 jar，最后将 JobGraph 提交到 YARN 上的 JobManager。 如果传递 --detached 参数，一旦提交被接受，客户端将停止

注意：部署 Per-Job 集群后，可以与其交互以执行取消或获取检查点等操作

① 直接提交 Job

./bin/flink run \
      -t yarn-per-job \
      -yjm 1024 \
      -ytm 1024 \
      -d \
      -c org.apache.flink.streaming.examples.socket.SocketWindowWordCount \
      examples/streaming/SocketWindowWordCount.jar \
      --port 9000

② 查看 yarn application 页面

③ 查看 WEB 界面

④ 列出运行的 Job

# 列出运行的 Job, 需要指定 application id
./bin/flink list \
      -t yarn-per-job \
      -Dyarn.application.id=application_1646190502615_0089

⑤ 取消运行的 Job

# 命令格式
./bin/flink cancel \
      -t yarn-per-job \
      -Dyarn.application.id=application_XXXX_YY <jobId>

# 示例
./bin/flink cancel \
      -t yarn-per-job \
      -Dyarn.application.id=application_1646190502615_0089 e31dd638510431fa579c7eaa9890961c

注意：取消 Job 后, 我们会发现, 整个 application 都停止了, 所以对应的集群也停止了