flink1.12.x：per-job cluster模式（yarn-cluster）部署

作者：马育民 • • 阅读：10446

准备

部署好 hadoop 集群

需要启动 yarn

依赖yarn运行，所以必须启动yarn命令如下：

/program/bin/hadoop.sh start

启动HDFS

如果需要读写HDFS，还需要启动HDFS

部署方式

在 Flink On YARN Per-Job 模式中：
Flink 只负责提交任务，不负责执行任务；
真正的 JobManager、TaskManager 全部由 YARN 动态启动容器运行，不需要提前部署。

所以：

  • 只需要任意一台能访问 YARN + HDFS 的机器
  • 解压 Flink、配置好 Hadoop 配置
  • 就能提交 Per-Job 任务
  • 不需要在所有 NodeManager 安装 Flink
  • 不需要在集群每台机器分发安装包

Per-Job 运行原理

Per-Job 的完整流程是这样的：

  1. 在某一台机器执行 flink run -m yarn-cluster
    这台机器上的 Flink 只是客户端（Client），作用只有：

    • 解析任务代码
    • 生成 JobGraph
    • 向 YARN ResourceManager 发起提交请求

  2. YARN 收到请求，在集群中找一台节点启动 AM
    AM = Flink JobManager
    这个过程是 YARN 自动调度的，不需要那台节点提前装 Flink。

  3. JobManager 向 YARN 申请 Container 启动 TaskManager
    Container 启动时，YARN 会自动把 Flink 的依赖包分发到对应节点。

  4. TaskManager 在对应节点上启动并执行任务
    节点不需要预装 Flink，完全是动态拉起、动态销毁。

  5. 任务结束 → AM 退出 → Container 释放 → 资源回收

整个过程：
只有客户端需要 Flink 安装包，计算节点完全不需要。

Flink 安装包只充当「客户端」

Flink On YARN 里有三个角色：

  1. Client（客户端）

    • 负责提交命令、打包任务、提交到 YARN
    • 这就是你那台安装 Flink 的机器
    • 只需要一台

  2. JobManager（ApplicationMaster）

    • 由 YARN 自动启动
    • 所在节点不需要部署 Flink

  3. TaskManager

    • 由 YARN 在任意节点启动容器
    • 所在节点同样不需要安装 Flink

所以从架构上就决定了：
只需要一台客户端机器，就能驱动整个 YARN 集群运行 Flink 任务。

企业生产部署标准实践

在真实公司大数据平台中：

  • 通常只在 网关机器（Gateway） 或 某一台运维节点 安装 Flink
  • 所有用户通过这台机器提交任务
  • YARN 负责在整个集群调度资源
  • 计算节点完全不安装、不维护 Flink

这就是：
一处部署，全局调度。

登录 hadoop1

登录 hadoop1 ，执行下面操作

上传

将 flink-1.12.5-bin-scala_2.12.tgz 上传到 hadoop1 的 /program 目录

解压缩

root 用户执行下面命令：

tar zxvf flink-1.12.5-bin-scala_2.12.tgz --no-same-owner

配置环境变量

修改 bigdata_env.sh

vim /etc/profile.d/bigdata_env.sh

增加 hadoop 配置目录

增加下面内容：

export HADOOP_CONF_DIR=/program/hadoop-3.0.3/etc/hadoop

必须加上下面环境变量：

export HADOOP_CLASSPATH=`hadoop classpath`

否则报错如下：

增加 FLINK_HOME

# 配置 FLINK_HOME
export FLINK_HOME=/program/flink-1.12.5
export PATH=${FLINK_HOME}/bin:$PATH

立即生效

source /etc/profile

修改 conf/flink-conf.yaml

注意： 是 yaml 文件，:后面要有空格

vim conf/flink-conf.yaml

taskmanager.numberOfTaskSlots

找到下面配置，修改如下：

taskmanager.numberOfTaskSlots: 2

相当于 taskmanager 有 2 个线程执行

注意：要根据 yarn 资源配置。该值不要轻易改大，否则可能会因为虚拟机配置低，导致 提交 flink 任务时，无法申请 slots

修改（可略）

在 flink 配置文件里 flink-conf.yaml设置

classloader.check-leaked-classloader: false

可能报错：

Exception in thread "Thread-7" java.lang.IllegalStateException: Trying to access closed classloader. Please check if you s              tore classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third part              y library and cannot be fixed immediately, you can disable this check with the configuration 'classloader.check-leaked-cla              ssloader'.

测试

在 flink 的 examples 目录下，都是一些官方的例子，如：WordCount.jar，读取一个文件，计算 wordcount，然后将结果保存到另一个文件中

创建测试文件

创建文本文件，内容如下：

李雷 lucy
韩梅梅 lili
lucy 李雷
lili 张三
张三
李四
王五
李雷
韩梅梅
lucy
李雷
韩梅梅
李雷

将其上传到 hdfs 的 /test2 目录下

提交任务

如下：

flink run -m yarn-cluster /program/flink-1.12.5/examples/batch/WordCount.jar --input hdfs://hadoop1:8020/test2/data.txt --output hdfs://hadoop1:8020/flink_result7

解释：

  • -m：指定提交到哪台主机上
  • yarn-cluster：表示提交到 yarn 集群上
  • --input：要读取的文件路径
  • --output：要保存结果的文件

注意： hdfs://hadoop1:8020/ 中的 flink_result7 文件会自动创建，如果已经存在，会报错！

提示：此案例可能只统计英文，因为过滤非英文了

原文出处：/show_1IX2OC9eHGnH.html

吉ICP备15003043号