Flink on YARN（下）：常见问题与排查思路

2022-07-24 21:15:56

作者：杨弢（搏远）

Flink 支持 Standalone 独立部署和 YARN、Kubernetes、Mesos 等集群部署模式，其中 YARN 集群部署模式在国内的应用越来越广泛。Flink 社区将推出 Flink on YARN 应用解读系列文章，分为上、下两篇。上篇分享了基于 FLIP-6 重构后的资源调度模型介绍 Flink on YARN 应用启动全流程，本文将根据社区大群反馈，解答客户端和 Flink Cluster 的常见问题，分享相关问题的排查思路。

客户端常见问题与排查思路

▼ 应用提交控制台异常信息：Could not build the program from JAR file.

这个问题的迷惑性较大，很多时候并非指定运行的 JAR 文件问题，而是提交过程中发生了异常，需要根据日志信息进一步排查。最常见原因是未将依赖的 Hadoop JAR 文件加到 CLASSPATH，找不到依赖类（例如：ClassNotFoundException: org.apache.hadoop.yarn.exceptions.YarnException）导致加载客户端入口类（FlinkYarnSessionCli)失败。

**▼ Flink on YARN 应用提交时如何关联到指定 YARN 集群？
**

Flink on YARN 客户端通常需配置 HADOOP_CONF_DIR 和 HADOOP_CLASSPATH 两个环境变量来让客户端能加载到 Hadoop 配置和依赖 JAR 文件。示例（已有环境变量 HADOOP_HOME 指定 Hadoop 部署目录）：

export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop
export HADOOP_CLASSPATH=`${HADOOP_HOME}/bin/hadoop classpath`

▼ 客户端日志在哪里，如何配置？

客户端日志通常在 Flink 部署目录的 log 文件夹下：${FLINK_HOME}/log/flink-${USER}-client-.log，使用 log4j 配置：${FLINK_HOME}/conf/log4j-cli.properties。

有的客户端环境比较复杂，难以定位日志位置和配置时，可以通过以下环境变量配置打开 log4j 的 DEBUG 日志，跟踪 log4j 的初始化和详细加载流程：export JVM_ARGS="-Dlog4j.debug=true"

▼ 客户端疑难问题排查思路

当客户端日志无法正常定位时，可以修改 log4j 配置文件将日志级别由 INFO 改为 DEBUG 后重新运行，看是否有 DEBUG 日志可以帮助排查问题。对于一些没有日志或日志信息不完整的问题，可能需要开展代码级调试，修改源码重新打包替换的方式太过繁琐，推荐使用 Java 字节码注入工具 Byteman（详细语法说明请参考：Byteman Document），使用示例：

(1) 编写调试脚本，例如打印 Flink 实际使用的 Client 类，以下脚本表示在 CliFrontend#getActiveCustomCommandLine 函数退出时打印其返回值；

RULE test
CLASS org.apache.flink.client.cli.CliFrontend
METHOD getActiveCustomCommandLine
AT EXIT
IF TRUE
DO traceln("------->CliFrontend#getActiveCustomCommandLine return: "+$!);
ENDRULE

(2) 设置环境变量，使用 byteman javaagent：

export BYTEMAN_HOME=/path/to/byte-home
export TRACE_SCRIPT=/path/to/script
export JVM_ARGS="-javaagent:${BYTEMAN_HOME}/lib/byteman.jar=script:${TRACE_SCRIPT}"

(3) 运行测试命令 bin/flink run -m yarn-cluster -p 1 ./examples/streaming/WordCount.jar ，控制台将输出内容：

------->CliFrontend#getActiveCustomCommandLine return: org.apache.flink.yarn.cli.FlinkYarnSessionCli@25ce9dc4

Flink Cluster 常见问题与排查思路

▼ 用户应用和框架 JAR 包版本冲突问题

该问题通常会抛出 NoSuchMethodError/ClassNotFoundException/IncompatibleClassChangeError 等异常，要解决此类问题：

**
1.首先需要根据异常类定位依赖库*，然后可以在项目中执行 mvn dependency:tree 以树形结构展示全部依赖链，再从中定位冲突的依赖库，也可以增加参数 -Dincludes 指定要显示的包，格式为 [groupId]:[artifactId]:[type]:[version]，支持匹配，多个用逗号分隔，例如：mvn dependency:tree -Dincludes=power,javaassist；

2.定位冲突包后就要考虑如何排包，简单的方案是用 exclusion 来排除掉其从他依赖项目中传递过来的依赖，不过有的应用场景需要多版本共存，不同组件依赖不同版本，就要考虑用 Maven Shade 插件来解决，详情请参考 Maven Shade Plugin。

▼ 依赖库有多版本 JAR 包共存时如何确定某类的具体来源？

很多应用运行 CLASSPATH 中存在相同依赖库的多个版本 JAR 包，导致实际使用的版本跟加载顺序有关，排查问题时经常需要确定某个类的来源 JAR，Flink 支持给 JM/TM 进程配置 JVM 参数，因此可以通过下面三个配置项来打印加载类及其来源（输出在 .out 日志），根据具体需要选择其中之一即可：

env.java.opts=-verbose:class   //配置JobManager&TaskManager
env.java.opts.jobmanager=-verbose:class  //配置JobManager
env.java.opts.taskmanager=-verbose:class  //配置TaskManager

▼ Flink 应用的完整日志如何查看？

Flink 应用运行中的 JM/TM 日志可以在 WebUI 上查看，但是查问题时通常需要结合完整日志来分析排查，因此就需要了解 YARN 的日志保存机制，YARN 上 Container 日志保存位置跟应用状态有关：

1.如果应用还没有结束，Container 日志会一直保留在其运行所在的节点上，即使 Container 已经运行完成仍然可以在所在节点的配置目录下找到： ${yarn.nodemanager.log-dirs}//，也可以直接从 WebUI访问：http:///node/containerlogs//

2.如果应用已结束并且集群启用了日志收集(yarn.log-aggregation-enable=true)，则通常应用结束后（也有配置可以增量上传）NM会将其全部日志上传至分布式存储（通常是 HDFS）并删除本地文件，我们可以通过 yarn 命令 yarn logs -applicationId -appOwner 查看应用的全部日志，还可以增加参数项 -containerId -nodeAddress 来查看某 container 的日志，也可以直接访问分布式存储目录：${yarn.nodemanager.remote-app-log-dir}/${user}/${yarn.nodemanager.remote-app-log-dir-suffix}/

▼ Flink 应用资源分配问题排查思路

如果 Flink 应用不能正常启动达到 RUNNING 状态，可以按以下步骤进行排查：

1.需要先检查应用当前状态，根据上述对启动流程的说明，我们知道：

处于 NEW_SAVING 状态时正在进行应用信息持久化，如果持续处于这个状态我们需要检查 RM 状态存储服务（通常是 ZooKeeper 集群）是否正常；
如果处于 SUBMITTED 状态，可能是 RM 内部发生一些 hold 读写锁的耗时操作导致事件堆积，需要根据 YARN 集群日志进一步定位；
如果处于 ACCEPTED 状态，需要先检查 AM 是否正常，跳转到步骤2；
如果已经是 RUNNING 状态，但是资源没有全部拿到导致 JOB 无法正常运行，跳转到步骤 3；

2.检查 AM 是否正常，可以从 YARN 应用展示界面（http:///cluster/app/）或 YARN 应用 REST API（http:///ws/v1/cluster/apps/）查看 diagnostics 信息，根据关键字信息明确问题原因与解决方案：

- Queue's AM resource limit exceeded. 原因是达到了队列 AM 可用资源上限，即队列的 AM 已使用资源和 AM 新申请资源之和超出了队列的 AM 资源上限，可以适当调整队列 AM 可用资源百分比的配置项：yarn.scheduler.capacity..maximum-am-resource-percent。

- User's AM resource limit exceeded. 原因是达到了应用所属用户在该队列的 AM 可用资源上限，即应用所属用户在该队列的 AM 已使用资源和 AM 新申请资源之和超出了应用所属用户在该队列的 AM 资源上限，可以适当提高用户可用 AM 资源比例来解决该问题，相关配置项：yarn.scheduler.capacity..user-limit-factor 与 yarn.scheduler.capacity..minimum-user-limit-percent。

- AM container is launched, waiting for AM container to Register with RM. 大致原因是 AM 已启动，但内部初始化未完成，可能有 ZK 连接超时等问题，具体原因需排查 AM 日志，根据具体问题来解决。

- Application is Activated, waiting for resources to be assigned for AM. 该信息表示应用 AM 检查已经通过，正在等待调度器分配，此时需要进行调度器层面的资源检查，跳转到步骤 4。

3.确认应用确实有 YARN 未能满足的资源请求：从应用列表页点击问题应用 ID 进入应用页面，再点击下方列表的应用实例 ID 进入应用实例页面，看 Total Outstanding Resource Requests 列表中是否有 Pending 资源，如果没有，说明 YARN 已分配完毕，退出该检查流程，转去检查 AM；如果有，说明调度器未能完成分配，跳转到步骤4；

4.调度器分配问题排查，YARN-9050 支持在 WebUI 上或通过 REST API 自动诊断应用问题，将在 Hadoop3.3.0 发布，之前的版本仍需进行人工排查：

检查集群或 queue 资源，scheduler 页面树状图叶子队列展开查看资源信息：Effective Max Resource、Used Resources：（1）检查集群资源或所在队列资源或其父队列资源是否已用完；（2）检查叶子队列某维度资源是否接近或达到上限；
检查是否存在资源碎片：（1）检查集群 Used 资源和 Reserved 资源之和占总资源的比例，当集群资源接近用满时（例如 90% 以上），可能存在资源碎片的情况，应用的分配速度就会受影响变慢，因为大部分机器都没有资源了，机器可用资源不足会被 reserve，reserved 资源达到一定规模后可能导致大部分机器资源被锁定，后续分配可能就会变慢；（2）检查 NM 可用资源分布情况，即使集群资源使用率不高，也有可能是因为各维度资源分布不同造成，例如 1/2 节点上的内存资源接近用满 CPU 资源剩余较多，1/2 节点上的 CPU 资源接近用满内存资源剩余较多，申请资源中某一维度资源值配置过大也可能造成无法申请到资源；
检查是否有高优先级的问题应用频繁申请并立即释放资源的问题，这种情况会造成调度器忙于满足这一个应用的资源请求而无暇顾及其他应用；
检查是否存在 Container 启动失败或刚启动就自动退出的情况，可以查看 Container 日志(包括 localize 日志、launch 日志等)、YARN NM 日志或 YARN RM 日志进行排查。

▼ TaskManager 启动异常：
org.apache.hadoop.yarn.exceptions.YarnException: Unauthorized request to start container. This token is expired. current time is ... found ...

该异常在 Flink AM 向 YARN NM 申请启动 token 已超时的 Container 时抛出，通常原因是 Flink AM 从 YARN RM 收到这个 Container 很久之后（超过了 Container有效时间，默认 10 分钟，该 Container 已经被释放）才去启动它，进一步原因是 Flink 内部在收到 YARN RM 返回的 Container 资源后串行启动。

当待启动的 Container 数量较多且分布式文件存储如 HDFS 性能较慢（启动前需上传 TaskManager 配置）时 Container 启动请求容易堆积在内部，FLINK-13184 对这个问题进行了优化，一是在启动前增加了有效性检查，避免了无意义的配置上传流程，二是进行了异步多线程优化，加快启动速度。

▼ Failover 异常 1：
java.util.concurrent.TimeoutException: Slot allocation request timed out for ...

异常原因是申请的 TaskManager 资源无法正常分配，可以按 Flink 应用资源分配问题排查思路的步骤4排查问题。

▼ Failover 异常 2：
java.util.concurrent.TimeoutException: Heartbeat of TaskManager with id timed out.

异常直接原因是 TaskManager 心跳超时，进一步原因可能有：

进程已退出，可能自身发生错误，或者受到 YARN RM 或 NM 上抢占机制影响，需要进一步追查 TaskManager 日志或 YARN RM/NM 日志；
进程仍在运行，集群网络问题造成失联，连接超时会自行退出，JobManager 在该异常后会 Failover 自行恢复（重新申请资源并启动新的 TaskManager）；
进程 GC 时间过长，可能是内存泄露或内存资源配置不合理造成，需根据日志或分析内存进一步定位具体原因。

▼ Failover 异常 3：
java.lang.Exception: Container released on a lost node

异常原因是 Container 运行所在节点在 YARN 集群中被标记为 LOST，该节点上的所有 Container 都将被 YARN RM 主动释放并通知 AM，JobManager 收到此异常后会 Failover 自行恢复（重新申请资源并启动新的 TaskManager），遗留的 TaskManager 进程可在超时后自行退出。

▼ Flink Cluster 疑难问题排查思路

首先根据 JobManager/TaskManager 日志分析定位问题，完整日志请参考“Flink 应用的完整日志如何查看”，如果想获取 DEBUG 信息，需修改 JobManager/TaskManager 的 log4j 配置（${FLINK_HOME}/conf/log4j.properties）后重新提交运行，对于仍在运行的进程，推荐使用 Java 字节码注入工具 Byteman 来一窥进程内部的相关状态，详细说明请参考：How Do I Install The Agent Into A Running Program?