简介面向CDH6.3.2集群集成Flink1.16.2的部署场景其中parcel、csd、依赖jar等文件已一次性打包齐全可以让大数据平台工程师、集群运维人员省去自行编译或组装安装包的繁琐工作直接进入集成配置环节。压缩包内共7个文件以4个jar包和1个parcel安装包为主体辅以sha校验文件与manifest.json描述文件其中parcel用于CDH直接识别分发csd jar用于Cloudera Manager服务管理shaded依赖jar提供Hadoop 3.1.1兼容支持各文件职责明确总大小约518.32MB。资源同时附上两篇部署教程链接覆盖parcel上传激活、服务配置及Flink on YARN运行模式等关键环节也给出集成过程中的常见问题处理思路可有效减少反复试错成本。目前已有972人学习下载适合正在搭建或升级CDH6.3.2Flink1.16.2环境的读者直接使用也可作为团队内部实施参考。1. Flink 1.16.2 与 CDH 6.3.2CDH 老集群上跑流计算的最终解法CDH 6.3.2 的用户想升级 Flink 到 1.16.2最大的障碍不是 Flink 本身而是官方的 parcel 仓库已经不再为 CDH 6.3.2 提供新版本支持。上周接到一个某公司的紧急任务生产环境还是 CDH 6.3.2但 Flink 停留在 1.14 时代多处 SQL 语法和窗口行为在 1.14 上有已知问题需要平滑升级到 Flink 1.16.2。官方没有对应的 parcel自己从源码编译耗时且容易踩坑。这份 Flink-1.16.2 CDH6.3.2 parcels 资源正好补齐了这个缺口——打好包的 parcel 文件和对应 CSD通过 CM 的本地仓库机制分发到集群的所有节点十分钟左右就能完成 Flink 运行时的整体替换。适合 CDH 6.3.2 且没有独立 Flink 集群、希望直接复用 YARN 资源的管理员。2. 为什么必须要用 parcel对比二进制包和源码编译的三个差距2.1 CDH 集群分发机制的底层逻辑CDH 用 Cloudera Manager 管理集群节点上的组件生命周期都归 CM 管。CM 的分发单元就是 parcel——一个按特定目录结构打包的 tar.gz 文件里面有 manifest.json 记录版本和依赖关系还有专用的校验签名。CM 拿到 parcel 后会分发到每个主机的/opt/cloudera/parcel-cache然后解压到/opt/cloudera/parcels并做软链。这套机制带来的直接好处是分发、激活、回滚都有标准路径不会出现不同节点 Flink 版本不一致的情况。对比直接往/opt丢一个 Flink 二进制包parcel 方式能复用 CM 的主机列表和故障处理逻辑。二进制包需要自己写脚本同步到所有节点还要处理节点间库版本不一致的问题。而源码编译更痛苦Flink 1.16.2 编译要拉一堆依赖Scala 2.12 和 Hadoop 3/CDH 的兼容性问题能折腾一整天。2.2 三个关键差异版本管理、依赖隔离、回滚机制版本管理。parcel 里有版本号CM 的 parcel 管理界面能看到每台主机的激活状态。二进制包版本号在目录名里换版本要手动改 PATH 和软链。依赖隔离。parcel 自带的 lib 目录把 Flink 运行所需的 Hadoop、YARN 相关 jar 隔离在自身目录内不会污染系统目录。二进制包解压后虽然有 lib但缺少和 CM 通信的配置经常出现 Flink 任务里读不到 HDFS 的 core-site.xml 的问题。回滚机制。CM 对 parcel 提供取消激活功能一键退回旧版本。二进制包的回滚是自己保留旧目录、改软链出错时恢复流程全靠记忆。Flink 升级这种事敢没有后悔药的不多。2.3 什么时候不该用 parcel我之前在某个测试环境试用过一套 Flink 1.17.0 的 CDH parcel遇到严重问题该 parcel 是直接对 Apache Flink 的发布包重新打包的但 CDH 6.3.2 的 YARN 资源管理器 API 与 Flink 1.17.0 的期望行为不一致导致作业在 YARN 上反复申请不到容器。后来我意识到CDH 6.3.2 这种 2020 年左右的技术栈应该匹配 2023 年初的 Flink 1.16.2这套搭配才避开了后续 YARN 接口变化的坑。所以先确认你的 CDH 版本再选择对应 Flink 版本CDH 6.3.2 就用 1.16.x不要贪新版。3. 整套接入流程下载、建本地仓库、分发、激活、部署 CSD3.1 资源清单与本地仓库建立这套资源里包含四个文件Flink-1.16.2 的 parcel 文件、对应的 sha 校验文件、manifest.json、以及 Flink 的 CSD 文件jar 形式。拿到后先在 CM 服务器上建本地仓库目录让 CM 能通过 HTTP 访问到这些文件。我习惯把 parcel 仓库放在 CM 自带的 Apache HTTPD 目录下这样不用额外开服务。仓库路径按下面的方式准备# 在 CM 服务器上执行路径可根据你的 CM 部署方式调整 mkdir -p /var/www/html/flink-parcels cp Flink-1.16.2-cdh6.3.2-parcel.jar /var/www/html/flink-parcels/ cp Flink-1.16.2-cdh6.3.2.parcel /var/www/html/flink-parcels/ cp Flink-1.16.2-cdh6.3.2.parcel.sha /var/www/html/flink-parcels/ cp manifest.json /var/www/html/flink-parcels/ cp FLINK-1.16.2.jar /opt/cloudera/csd/ # 验证 HTTP 访问 curl -I http://localhost/flink-parcels/manifest.json关键参数说明FLINK-1.16.2.jar是 CSD 文件决定 CM 界面里 Flink 服务的显示名称和角色划分.sha文件是 parcel 的 SHA-1 值CM 校验用不能用 MD5 替代manifest.json里记录了 parcel 的版本号、名称和依赖CM 读取仓库目录时的入口就是它。如果 HTTP 访问返回 404检查 SELinux 是否拦截了 HTTPD 的目录访问# 临时关闭 SELinux 或放行 HTTPD 对 /var/www/html 的读写 setsebool -P httpd_read_user_content 13.2 在 CM 界面配置本地仓库CM 界面的操作路径很固定管理 - 设置 - 外部资源 - Parcel 存储库。把默认的仓库 URL 后面加上本机地址格式是http://cm-server-host/flink-parcels。注意把本地仓库地址追加在默认 URL 后面用逗号分隔不要替换掉原来的默认仓库。原因在于CDH 自身的 parcel比如 Spark、Kafka还需要从默认仓库拉取替换后可能导致其他组件无法继续分发。配置完后等待 CM 刷新仓库列表然后进入Parcel - 可用页面应该能看到 Flink-1.16.2。如果看不到点击检查新 parcel按钮手动触发重新读取。还有一点manifest.json里如果声明了依赖其他 parcel比如cdh那 CDH 本身的 parcel 也必须在已下载状态否则 Flink 的 parcel 无法分配。3.3 分配、激活与 CSD 部署在 Parcel 页面找到 Flink 1.16.2先点下载下载完成后点分配CM 会把 parcel 推送到集群内所有主机的/opt/cloudera/parcel-cache。分配完成后点激活CM 在所有节点上解压并建立软链。激活完成之后回到 CM 首页Flink 服务应该出现在服务列表里。但这里有个常见跨版本陷阱如果资源里的 CSD jar 版本与 parcel 版本不严格匹配CM 界面会报“服务不可用”或者压根不显示 Flink 服务。CSD 必须和 parcel 配套使用不能拿老版本 CSD 驱动新版本 parcel。# 检查各节点的软链是否正确 ls -l /opt/cloudera/parcels/flink # 预期输出是软链到具体版本目录 # flink - /opt/cloudera/parcels/FLINK-1.16.2/如果软链指向的目录不存在或者 manifest 里写的目录名和实际解压目录不一致需要重新分配并激活。这种路径类的坑很难排查建议激活前确认 manifest 里的目录名称字段。我从第一次部署踩过坑后养成了习惯每个新 parcel 激活前先看 manifest.json 里的名字和版本号避免路径对不上。3.4 在 CM 中创建 Flink 服务与 YARN 集成配置激活后进入 CM 的实例 - 添加服务选择 Flink角色建议这样分配一个Flink Gateway角色用于提交作业和多个Flink History Server角色用于查看作业日志。Gateway 放在提交作业的节点上History Server 放在资源充足的节点上。关键配置项在配置 - 高级里需要修改这几个参数配置项推荐值说明yarn.application.classpath或 Flink 的flink.yarn.classpath使用 parcel 内嵌的 Hadoop 依赖不要引用系统 Hadoop避免与 CDH 自带 Hadoop 版本冲突history-server.archive.fs.dirhdfs://nameservice1/flink/completed-jobs/需要先在 HDFS 建好目录否则 History Server 起不来flink.hadoop.home/opt/cloudera/parcels/CDH/lib/hadoop指向 CDH parcel 内的 Hadoop 客户端# 在 HDFS 中创建作业归档目录 sudo -u hdfs hdfs dfs -mkdir -p /flink/completed-jobs sudo -u hdfs hdfs dfs -chown -R flink:flink /flink/completed-jobs这里flink.hadoop.home是很多环境跑不起来的核心原因。Flink 客户端在提交作业时需要找到 Hadoop 的配置文件core-site.xml、hdfs-site.xml、yarn-site.xml不配这个参数时 Flink 用的是自身捆绑的 Hadoop 配置里面是空配置根本连不上 HDFS。最好通过HADOOP_CONF_DIR环境变量显式指定配置目录让 Flink 作业能直接拿到 HDFS 和 YARN 的地址。3.5 第一个 Flink 作业的提交验证配置完成后建议在 Gateway 节点上先跑一个最简单的作业验证链路不要一上来就提交复杂 SQL# 找一个测试 jar比如 flink 自带的 wordcount 示例 /opt/cloudera/parcels/flink/bin/flink run \ -t yarn-per-job \ -m yarn-cluster \ -ynm flink-wordcount-test \ -yjm 1024m \ -ytm 1024m \ -ys 1 \ /opt/cloudera/parcels/flink/examples/batch/WordCount.jar \ --input hdfs:///tmp/input.txt --output hdfs:///tmp/output.txt参数说明-t yarn-per-job指定在 YARN 上以 per-job 模式运行每个作业起一个独立集群-ynm是 YARN 上应用的名字用于在 Resource Manager 界面识别-yjm和-ytm分别是 JobManager 和 TaskManager 的内存-ys是每个 TaskManager 的 slot 数。跑这个命令前先确认/tmp/input.txt在 HDFS 上存在否则作业提交成功但运行时直接报错“Input path does not exist”。如果作业在 YARN 上提交后迅速失败第一件事是看 Resource Manager 日志重点找ApplicationMaster相关的报错# 查看 YARN 上 flink 应用的日志 yarn logs -applicationId application_xxx日志里出现ClassNotFoundException时说明 Flink 作业缺少 CDH 的依赖 jar出现InvalidConfigurationException时优先检查yarn-site.xml是否被正确加载。4. 避坑指南从 parcel 校验到作业运行时的五个高频故障4.1 坑一parcel 校验失败提示 SHA-1 不匹配下载完 parcel 后在 CM 界面点分配直接报错parcel 校验失败,expected sha1 ... but got sha1 ...。原因出在.sha文件内容不对可能是复制时把.sha文件里的实际哈希值弄错了或者从网络下载的.sha文件本身对应的是另一个版本的 parcel。解决方法是本地重新计算 parcel 的 SHA-1 值然后替换.sha文件内容# 重新生成 sha1 校验值注意格式后面的空格和文件名不能省略 sha1sum Flink-1.16.2-cdh6.3.2.parcel Flink-1.16.2-cdh6.3.2.parcel.sha # 查看生成的 .sha 内容 cat Flink-1.16.2-cdh6.3.2.parcel.sha # 输出格式类似a3f9c2b6e63b42a7e7f925a57a1f2b1e51f68c2b Flink-1.16.2-cdh6.3.2.parcel注意.sha文件内容格式必须严密哈希值 空格 文件名。有些从网上下载的.sha文件只有哈希值没有文件名CM 也会报校验失败。从那以后我换了做法每次拿到 parcel 资源先手动执行一次sha1sum重新生成.sha文件避免各种传输过程中文件内容被篡改或格式不对的问题。另一个细节如果读者使用 Windows 下载文件再传到 Linuxunix2dos转换或 UTF-8 BOM 头都会破坏.sha文件的格式务必用file命令检查文件编码。4.2 坑二CSD 加载失败CM 界面看不到 Flink 服务CSD jar 复制到/opt/cloudera/csd后重启 CM Server界面里没有出现 Flink 服务或出现但报错“CSD 解析失败”。原因一般是 CSD jar 版本和 CM 版本不兼容或者 jar 包内容损坏、缺少描述文件。排查思路先确认 CM 版本CSD 需要匹配 CM 的主版本。然后再确认 jar 不是损坏状态# 检查 jar 包内描述文件是否存在 unzip -l FLINK-1.16.2.jar | grep -i descriptor # 正常输出里包含 flink-csd.xml 之类的描述文件如果 CSD jar 没有问题检查 CM 日志# 日志路径一般在 /var/log/cloudera-scm-server/ grep -i csd\|flink /var/log/cloudera-scm-server/cloudera-scm-server.log | tail -200日志里如果出现Failed to parse ...的行说明 CSD 的 XML 描述文件里引用了 CM 不支持的 service 类型或命令名称。这种只能换 CSD 版本没有其他办法。还有一点可能被忽略CSD jar 文件名要和 manifest 里声明的服务名对应比如 jar 命名为FLINK-1.16.2.jar里面的服务名是FLINK大小写不敏感但拼写必须一致。4.3 坑三YARN 作业提交报错Unsupported Hadoop versionFlink 作业在 CM 的 Gateway 节点提交时报错Unsupported Hadoop version x.y.z或者直接抛NoClassDefFoundError: org/apache/hadoop/yarn/api/records/ApplicationId。原因是 Flink 1.16.2 默认的 Hadoop 依赖版本Hadoop 2.10.2 左右与 CDH 6.3.2 内置的 Hadoop 3.0.0 不兼容。CDH 6.3.2 用的 Hadoop 3.0.0-cdh6.3.2YARN ApplicationId 的类定义在 3.0 里有了变化Flink 客户端在编译期绑定的还是老版本的类运行时遇到新类就炸。解决思路是在提交作业前让 Flink 客户端优先使用 CDH 的 Hadoop 客户端 jar而不是 Flink 自带的# 在 flink.conf 的 flink-conf.yaml 中追加 env.java.opts: -Djava.library.path/opt/cloudera/parcels/CDH/lib/hadoop/lib/native # 设置 HADOOP_CONF_DIR 指向 CDH 的配置目录 export HADOOP_CONF_DIR/etc/hadoop/conf如果仍然报错考虑用-Dflink.yarn.jars参数显式指定依赖 jar 的位置把 CDH 的核心 jar 加进去。经验是 Flink 1.16.2 搭配 CDH 6.3.2 时先在 Gateway 节点上用hadoop classpath拿到完整类路径然后对比 Flink 的lib目录把缺失的 CDH 专属 jar 复制到 Flink 的lib目录下。但要注意不要全量替换 Flink 的 lib只补 CDH 有而 Flink 没有的类。这个问题的根源是 Flink 的flink-shaded-hadoop模块在 1.16 版本后已经把 Hadoop 依赖打进了 Flink 核心包默认用的 Hadoop 版本偏低。后面我选择直接重新编译 Flink 或替换flink-shaded-hadoop的版本但风险更高。使用 parcel 内置的依赖并显式声明HADOOP_CONF_DIR是这个场景下最稳妥的做法。4.4 坑四History Server 启动后无法读取日志界面显示 404History Server 启动成功但访问 Web 界面时出现 404或者能打开但显示“No completed jobs”。日志里有java.io.FileNotFoundException: Failed to open file ... /flink/completed-jobs/_tmp...。原因通常是两部分作业完成后写入completed-jobs目录的文件权限不对或者history-server.archive.fs.dir配置的 HDFS 路径没建好。Flink 历史服务是通过flink.hadoop.home指向的 Hadoop 客户端访问 HDFS如果客户端使用的用户权限不足连目录都读不到。# 确认 HDFS 目录权限/flink 目录必须能让 flink 用户读写 sudo -u hdfs hdfs dfs -ls /flink # 如果权限不对重新设置 sudo -u hdfs hdfs dfs -chmod -R 755 /flink还有一个冷门原因Flink 作业配置了checkpoint.cleaner但没配置state.savepoints.dir作业正常结束后状态文件写不进去History Server 扫描时认为作业未完成。所以提交作业时最好显式指定# flink-conf.yaml 中追加 state.savepoints.dir: hdfs:///flink/savepoints4.5 坑五多个 Flink 作业互相抢占 TaskManager 资源在 YARN 上运行多个 Flink 作业时出现一个作业把资源占满后其他作业处于 PENDING 状态一直等不到资源。原因可能是yarn.scheduler.capacity默认配置里没有给 Flink 作业队列单独设置资源上限所有作业混在 default 队列里资源少的作业被饿死。YARN 上跑 Flink 作业时建议通过-yqu指定队列/opt/cloudera/parcels/flink/bin/flink run \ -t yarn-per-job \ -m yarn-cluster \ -yqu flink_queue \ -ys 2 \ -yjm 2048m -ytm 4096m \ /path/to/your-job.jar如果集群里没有flink_queue需要先在 Capacity Scheduler 里配置好队列给 Flink 队列留出固定比例资源。这个坑不是 parcel 本身的问题但属于混用 CDH 6.3.2 和 Flink 1.16.2 后最容易因为资源分配而引发的故障值得提前排查 YARN 队列配置。5. 提升资源利用率核心参数调优与多作业隔离5.1 内存与并行度参数的推荐配置CDH 6.3.2 集群一般是大数据平台的主要资源池Flink 作业跑在上面时内存参数要结合 YARN 的容器规格制定。Flink 1.16.2 的yarn-per-job模式下每个作业单独的 JobManager 和 TaskManager 都占用独立容器内存配置不能太随意。以常见的 16GB 物理机节点为例我一般这样配每台 TaskManager 分配 8GB 堆内存对应容器内存-ytm 10240m任务并行度按每个 slot 2GB 内存计算。flink-conf.yaml里的具体参数如下jobmanager.memory.process.size: 2048m taskmanager.memory.process.size: 10240m taskmanager.memory.managed.fraction: 0.4 taskmanager.numberOfTaskSlots: 4 parallelism.default: 1参数逻辑taskmanager.memory.process.size是整个 TaskManager 进程的总内存包含堆外、网络缓冲和 DirectMemorytaskmanager.memory.managed.fraction是留给 Flink 托管内存的比例用于窗口计算和排序。在 CDH 上如果这个值设得太大留给堆内执行的内存就变少SQL 作业里多表 join 容易报 OOM设得太小窗口聚合容易频繁落盘。5.2 多作业隔离与队列上限的控制方法CDH 集群上同时跑批处理Hive/Spark和流处理Flink时Flink 作业必须限制在专门的 YARN 队列里。否则 Flink 作业的长时间运行特征会让短时间批处理作业排队。# 在 Capacity Scheduler 配置中增加 flink 队列 # 具体路径是 YARN 服务的 Capacity Scheduler 配置页 # 队列名 flink容量 30%最大容量 50%同时在提交作业时强制指定队列/opt/cloudera/parcels/flink/bin/flink run-application \ -t yarn-application \ -m yarn-cluster \ -yqu flink \ -D yarn.application.priority5 \ /path/to/your-streaming-job.jar这里-D yarn.application.priority5是设置 YARN 上作业的优先级保证流作业和批作业同时提交时流作业优先被调度。5.3 使用 Application 模式替代 per-job 模式Flink 1.16.2 在 YARN 上支持yarn-application模式多个作业共享同一个 JobManager相比per-job模式每个作业启动独立 AM能显著减少资源浪费。CDH 6.3.2 的 YARN 版本支持这个模式切换代价很小。/opt/cloudera/parcels/flink/bin/flink run-application \ -t yarn-application \ -m yarn-cluster \ -p 20 \ -d \ /path/to/your-streaming-job.jar参数说明-p 20指定默认并行度-d表示 detached 模式提交后命令行立即返回。Application 模式对作业数量多有明显收益但不适合短时间运行的测试作业因为 JobManager 启动后要等第一个作业提交如果一直没作业提交JobManager 会一直空转消耗资源。如果有零星的临时任务保留per-job模式更划算。6. 升级后的验证与调试技巧作业稳定跑三天的完整检查单6.1 集群健康检查的四个维度升级完 Flink 1.16.2 后不要急着跑全量业务。先在非高峰期用真实数据做一次完整的健康检查我按四个维度来检查第一个维度是 Flink 作业的 Restart 次数。在 Flink Web UI 上打开作业详情看Restarting分支出现次数如果频繁重启说明作业在运行时遇到了周期性异常# 通过 REST API 快速获取作业异常的排查入口 curl http://localhost:8081/jobs/overview # 找到具体 job 的 id 后取异常详情 curl http://localhost:8081/jobs/job-id/exceptions第二个维度是 Checkpoint 的间隔和失败率。Flink 1.16.2 的 Web UI 里能看到 Checkpoint 的历史曲线连续失败超过三次说明状态后端比如 RocksDB和 HDFS 的连接有问题优先检查 HDFS 的写入性能和 NameNode 的 RPC 延迟。第三个维度是反压。进入作业的每个算子页面看背压状态是OK还是HIGH。只有某个算子的反压是HIGH优先调大并行度# 在 flink-conf.yaml 中调整 # 或者通过 SQL 客户端提交动态修改第四个维度是 YARN 资源利用率。确认各个队列的资源使用和剩余避免 FLink 作业占满 YARN 导致 Hive 作业没资源yarn top # 查看每个应用的资源占用情况6.2 使用日志定位作业失败toString 与异常栈的配合Flink 1.16.2 相比旧版本日志输出更丰富。但流作业的日志分散在 TaskManager 的 stdout 和 stderr 里直接从 Web UI 看很容易漏掉关键异常。我一般在提交命令里加上日志级别参数把 Flink 作业内部的日志单独输出到文件# 在 flink-conf.yaml 中调整日志级别 rootLogger.level: INFO logger.flink.name: org.apache.flink logger.flink.level: DEBUG然后通过 YARN 日志查看具体的异常yarn logs -applicationId application_xxx -containerId container_xxx | grep -A 10 Caused by另外如果遇到java.io.IOException: Too many open files这类错误不要在业务代码里找原因优先检查这台节点的ulimit -n限制。CDH 6.3.2 默认的进程文件数上限经常不够 Flink 任务使用特别是当 TaskManager 同时打开多个 HDFS 文件句柄时。6.3 升级后的一周观察期哪些指标要重点盯升级完成后的第一周是最容易暴露问题的时间窗口。我建议每天固定时间看这些数据Flink 作业的 Checkpoint 成功率、YARN 队列的整体利用率、History Server 的归档文件数量。在日志里加上作业 ID 和提交时间的标记方便在出问题时快速回查。在这套 CDH 6.3.2 Flink 1.16.2 环境上跑了一段时间后我形成一个习惯每次提交新作业前强制走一遍检查单——先确认 parcel 激活正常然后确认 YARN 队列有资源再配置好 HADOOP_CONF_DIR 和 classpath最后看一眼 HDFS 目录权限。这三分钟检查帮我过滤掉了至少一半的线上故障希望帮到你。本文还有配套的精品资源点击获取