0. 环境准备
本实验基于以下 8 台测试机器进行：
IP hostname 10.4.79.90 hadoop-1 10.4.79.8 hadoop-2 10.4.79.6 hadoop-3 10.4.79.58 hadoop-4 10.4.79.38 hadoop-5 10.4.79.96 hadoop-6 10.4.79.62 hadoop-7 10.4.79.92 hadoop-8
首先确认每个机器都安装了如下软件：
- JAVA >= 1.8.x
- SSH，并确保所有集群节点之间可互相 *SSH免密登录
为集群每个节点配置 hostname： vi /etc/hosts
10.4.79.90 hadoop-1
10.4.79.8 hadoop-2
10.4.79.6 hadoop-3
10.4.79.58 hadoop-4
10.4.79.38 hadoop-5
10.4.79.96 hadoop-6
10.4.79.62 hadoop-7
10.4.79.92 hadoop-8
YARN 环境搭建可参考 hadoop 环境搭建
1.1 准备
a. 测试 Yarn 环境是否已经准备好接受一个 Flink Application 了： yarn top
b. 下载 Flink 并解压缩
c. 系统环境添加配置项：
export HADOOP_CLASSPATH=hadoop classpath
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
1.2 以 Application 模式提交任务
./bin/flink run-application -t yarn-application ./examples/streaming/TopSpeedWindowing.jar
启动成功后，会返回一个 JobID，如下图所示：
启动日志中有一部分关键信息：
The configured JobManager memory is 1600 MB. YARN will allocate 2048 MB to make up an integer multiple of its minimum allocation memory (1024 MB, configured via
yarn.scheduler.minimum-allocation-mb). The extra 448 MB may not be used by Flink.
The configured TaskManager memory is 1728 MB. YARN will allocate 2048 MB to make up an integer multiple of its minimum allocation memory (1024 MB, configured via yarn.scheduler.minimum-allocation-mb). The extra 320 MB may not be used by Flink.
因为我们前文在 yarn-site.xml 中配置 yarn 的每个任务最小内存分配单元（yarn.scheduler.minimum-allocation-mb）是 1024MB，而我们在 Flink/conf/flink-conf.yaml 中配置的 jobmanager 内存使用大小（jobmanager.memory.process.size:）是 1600m，所以 yarn 会分配给 jobmanager 共计 2×1024=2048MB，Flink 只使用了其中的 1600MB，剩余 448MB，这些空间就浪费了。
同理，taskmanager 的内存大小配置是 1728MB，yarn 会分配给 taskmanager 共计 2×1024=2048MB，Flink 只使用了其中的 1728MB，剩余的 320MB 空间就浪费了。
因此我们将 Flink 的 JobManager (jobmanager.memory.process.size) 和 TaskManager (taskmanager.memory.process.size) 大小都配置为 2048M。
重新提交任务，启动成功后，可以在 yarn 管理页面看到正在运行的 Application：
点击任务 ID 跳转至任务详情页：
点击 ApplicationMaster 就跳转到了我们熟悉的 Flink 管理页面：
点击 AttempID 可以查看任务具体运行在哪个节点：
可以看到刚刚提交的 Flink 任务占用了两个 Yarn Container，分别在 hadoop-3 和 hadoop-7，一个 Container 用于运行 JobManager，一个 Container 用于运行 TaskManager。
1.3 日志查看
登陆到 hadoop-3 查看后台日志：
cd /opt/hadoop-2.10.0/logs/userlogs/{applicationid}/{containerid}
可以看出 hadoop-3 是一个 TaskManager。
登陆到 hadoop-7 查看后台日志：
cd /opt/hadoop-2.10.0/logs/userlogs/{applicationid}/{containerid}
可以看出 hadoop-7 是一个 JobManager。
1.4 日志归档
在 flink/conf 目录下已经提供了 log4j.properties 和 logback.xml，说明可以 log4j 和 logback 两种方案任选其一。
如果选择 logback 方案，需要移除（重命名）log4j.properties。
另外 官方文档 中还特别提示：
1.5 遇到的问题
a. 启动任务报错：
Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration ‘classloader.check-leaked-classloader’.
解决办法： 在 flink 配置文件 flink-conf.yaml 设置禁用 classloader.check-leaked-classloader
classloader.check-leaked-classloader: false
b. 无法提交新任务，状态一直处于 ACCEPTED：
Deployment took more than 60 seconds. Please check if the requested resources are available in the YARN cluster
似乎是资源不够用了。
- 查看 CPU 和内存资源占用情况Total Resources Used Resources
- 查看调度器的使用情况 可以看到，集群中使用的是
Capacity Scheduler调度器，也就是所谓的容量调度，这种方案更适合多租户安全地共享大型集群，以便在分配的容量限制下及时分配资源。采用队列的概念，任务提交到队列，队列可以设置资源的占比，并且支持层级队列、访问控制、用户限制、预定等等配置。但是，对于资源的分配占比调优需要更多的经验处理。但它不会出现在使用 FIFO Scheduler 时会出现的有大任务独占资源，会导致其他任务一直处于 pending 状态的问题。
- 解决验证 修改
hadoop/etc/hadoop/capacity-scheduler.xml：
yarn.scheduler.capacity.maximum-am-resource-percent
0.5
按照这个配置，最多可以运行 64GB x 0.5 / 2GB = 16 个 Flink Application。 修改完成后，刷新一下 yarn 队列：
yarn rmadmin -refreshQueues
继续提交任务，成功！
2. 参考
Original: https://www.cnblogs.com/myownswordsman/p/flink-on-yarn.html
Author: watermark’s
Title: Flink On YARN
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/564946/
转载文章受原作者版权保护。转载请注明原作者出处！