简介：这份资源面向需要在CDH 6.3.2集群上部署Flink 1.13.1的大数据开发与运维人员，解决Parcel分发、YARN资源调度与Flink配置衔接的实际问题。压缩包共5个文件，约299.52MB，包含两个jar（YARN客户端与Flink核心库）、一个parcel二进制发行版及其sha校验文件，以及描述版本依赖的manifest.json，覆盖从分发、校验到提交作业所需的完整组件。已有3496人学习下载，说明该组合在企业离线与实时数仓场景中具有较高参考价值。读者可据此完成Parcel导入与激活、flink-conf.yaml参数调整、YARN会话启动及作业提交，并借助Web UI或CDH管理界面监控运行状态，同时获得网络、权限与集群资源分配方面的排错思路，适合作为CDH集成Flink的落地参考。
1. 为什么 CDH 6.3.2 上跑 Flink 1.13.1 值得单独折腾一遍
如果你手头有一套 CDH 6.3.2 集群，又不想为了流计算单独维护一套原生 Apache Flink 集群，那 Flink 1.13.1 与 CDH 6.3.2 的集成包就是最省事的一条路。CDH 6.3.2 自带 Hadoop 3.0.0、Hive 2.1.1、HBase 2.1.0，而 Flink 1.13.1 恰好是对 Hadoop 3.x 兼容性比较稳的一个版本，两者拼在一起能直接复用 YARN 资源调度和 HDFS 存储，不用额外搭一套 Standalone 集群。这个组合适合已经在用 CDH 做离线数仓、现在想把实时链路也并进来的团队，也适合需要把 Flink SQL 跑在 Hive 数仓之上的场景。但要注意，CDH 官方并不提供 Flink 的 parcel，社区里流传的集成包质量参差不齐，选错一个包，后面 ClassNotFound 能把你折腾到怀疑人生。
2. 集成包选型与部署前必须搞清的依赖关系
2.1 为什么不能直接用 Apache 官方二进制包
很多人第一反应是去 Apache 官网下个flink-1.13.1-bin-scala_2.12.tgz，解压后往 CDH 节点上一扔就开跑。这么做在 Standalone 模式下确实能起来，但一旦切到 YARN 模式就会翻车。原因是 CDH 6.3.2 里的 Hadoop 是经过 Cloudera 重新打包的，部分依赖的 artifact 坐标和版本跟 Apache 原生 Hadoop 3.0.0 有差异，尤其是hadoop-common、hadoop-yarn-client这些包。Apache 官方 Flink 二进制包里捆绑的 Hadoop shaded jar 是按原生 Hadoop 编译的，跟 CDH 的类路径放一起，轻则警告，重则NoSuchMethodError。
常见做法是使用针对 CDH 重新编译的 Flink 集成包，这类包通常会把 Hadoop 依赖标记为provided，运行时直接复用 CDH 的 jar。判断一个集成包是否靠谱，看两点：一是它的编译目标 Hadoop 版本是否标注为 CDH 对应的 3.0.0-cdh6.3.2，二是解压后lib/目录下是否没有捆绑flink-shaded-hadoop-*.jar。
2.2 版本对齐清单
在动手之前，先把下面这张表里的版本关系确认一遍，任何一项对不上都可能导致运行时异常。
| 组件 | CDH 6.3.2 自带版本 | Flink 1.13.1 要求 | 注意事项 |
|---|---|---|---|
| Hadoop | 3.0.0-cdh6.3.2 | 3.0.0 及以上 | 必须用 CDH 编译的 Flink 包 |
| Hive | 2.1.1-cdh6.3.2 | 2.1.x 兼容 | Hive 连接器需单独放 jar |
| HBase | 2.1.0-cdh6.3.2 | 2.1.x 兼容 | 用 HBase 时才需要 |
| Scala | 2.11 / 2.12 | 2.11 或 2.12 | 与集成包编译版本一致 |
| JDK | 1.8 | 1.8 | 不支持 JDK 11 |
| Zookeeper | 3.4.5-cdh6.3.2 | 3.4+ | HA 模式依赖 |
这张表里最容易忽略的是 Scala 版本。Flink 1.13.1 同时提供 2.11 和 2.12 两个编译版本，你集群上如果已经有 Spark 用的是 Scala 2.11，那 Flink 最好也选 2.11，否则同一个 YARN 队列里两套 Scala 版本混跑，classpath 冲突是迟早的事。
2.3 部署路径规划
我一般把 Flink 集成包放在/opt/cloudera/parcels/之外的独立目录，比如/opt/flink-1.13.1，然后通过 Cloudera Manager 的「Gateway」配置或者手动分发的方式同步到所有 NodeManager 节点。这样做的好处是升级 Flink 不影响 CDH 本身的 parcel 管理，坏处是需要自己维护环境变量。
具体步骤：
# 1. 解压集成包到统一目录 tar -zxvf flink-1.13.1-cdh6.3.2-scala_2.12.tgz -C /opt/ mv /opt/flink-1.13.1 /opt/flink # 2. 设置环境变量，写入 /etc/profile.d/flink.sh export FLINK_HOME=/opt/flink export PATH=$FLINK_HOME/bin:$PATH # 3. 分发到所有 NodeManager 节点（假设有 node1/node2/node3） for host in node1 node2 node3; do scp -r /opt/flink $host:/opt/ scp /etc/profile.d/flink.sh $host:/etc/profile.d/ done解压后先别急着启动，检查lib/目录下有没有flink-shaded-hadoop-3-uber-*.jar这类文件。如果有，说明这个包是给原生 Hadoop 用的，需要手动移除或者替换成 CDH 版本的依赖。另外conf/flink-conf.yaml里的env.java.home建议显式指定为 CDH 使用的 JDK 路径，避免节点上多 JDK 版本导致启动脚本选错。
3. 核心配置：flink-conf.yaml 与 YARN 提交参数
3.1 flink-conf.yaml 关键项逐条拆解
集成包解压后，conf/flink-conf.yaml是核心配置文件。下面这几项是 CDH 环境下必须改的，其他保持默认即可。
# JobManager 和 TaskManager 的 RPC 地址，YARN 模式下由框架自动注入 jobmanager.rpc.address: localhost jobmanager.rpc.port: 6123 # 内存配置，根据 NodeManager 单节点可用内存调整 jobmanager.memory.process.size: 1600m taskmanager.memory.process.size: 4096m taskmanager.numberOfTaskSlots: 4 # YARN 相关，必须指向 CDH 的 Hadoop 配置目录 env.hadoop.conf.dir: /etc/hadoop/conf env.yarn.conf.dir: /etc/hadoop/conf # 检查点存储，CDH 环境建议直接写 HDFS 路径 state.backend: filesystem state.checkpoints.dir: hdfs://nameservice1/flink/checkpoints state.savepoints.dir: hdfs://nameservice1/flink/savepoints # 高可用，依赖 Zookeeper high-availability: zookeeper high-availability.zookeeper.quorum: node1:2181,node2:2181,node3:2181 high-availability.storageDir: hdfs://nameservice1/flink/ha high-availability.zookeeper.path.root: /flinkenv.hadoop.conf.dir和env.yarn.conf.dir这两项是 CDH 集成的关键。CDH 的 Hadoop 配置文件不在默认的$HADOOP_HOME/etc/hadoop下，而是由 Cloudera Manager 统一管理在/etc/hadoop/conf。如果不显式指定，Flink 提交到 YARN 时会找不到yarn-site.xml，报Cannot locate YARN configuration。
state.checkpoints.dir里的nameservice1要换成你集群实际的 HDFS nameservice ID，用hdfs getconf -confKey dfs.nameservices可以查到。如果集群没开 HA，直接写hdfs://node1:8020/flink/checkpoints。
3.2 提交任务到 YARN 的两种模式
Flink 1.13.1 支持per-job和application两种 YARN 提交模式。CDH 环境下我推荐用application模式，原因是 per-job 模式下每个任务都要起一个独立 JobManager，资源申请频繁，而 application 模式整个应用共用一个 JobManager，更适合长期运行的流任务。
# application 模式提交 $FLINK_HOME/bin/flink run-application \ -t yarn-application \ -Djobmanager.memory.process.size=1600m \ -Dtaskmanager.memory.process.size=4096m \ -Dtaskmanager.numberOfTaskSlots=4 \ -Dyarn.application.name=my-flink-job \ -c com.example.MyJob \ /path/to/my-job.jar # per-job 模式提交（对比用） $FLINK_HOME/bin/flink run \ -m yarn-cluster \ -yn 3 \ -ys 4 \ -c com.example.MyJob \ /path/to/my-job.jar-t yarn-application指定 application 模式，-D后面跟的是动态配置，会覆盖flink-conf.yaml里的同名项。-yn是 NodeManager 数量，-ys是每个 NodeManager 的 slot 数，这两个参数在 per-job 模式下用得多，application 模式下更推荐用-D方式统一管理。
提交后可以用yarn application -list查看应用状态，或者直接访问 YARN ResourceManager 的 Web UI。如果应用一直卡在ACCEPTED状态，多半是队列资源不足，检查yarn.scheduler.maximum-allocation-mb和队列的capacity配置。
3.3 Hive 集成配置
如果要用 Flink SQL 读 Hive 表，需要额外配置 Hive 连接器。CDH 6.3.2 的 Hive 是 2.1.1，Flink 1.13.1 对应的 Hive 连接器 jar 是flink-sql-connector-hive-2.1.1_2.12-1.13.1.jar，需要手动放到lib/目录下。
# 在 flink-conf.yaml 中追加 table.sql-dialect: hive table.catalog.default-database: default然后在 SQL Client 里注册 Hive Catalog：
CREATE CATALOG hive_catalog WITH ( 'type' = 'hive', 'hive-conf-dir' = '/etc/hive/conf', 'hadoop-conf-dir' = '/etc/hadoop/conf' ); USE CATALOG hive_catalog; SHOW DATABASES;hive-conf-dir指向 CDH 的 Hive 配置目录，里面要有hive-site.xml。如果报HiveConf相关错误，检查HADOOP_CLASSPATH是否包含了 Hive 的 jar 包，CDH 环境下通常需要手动 export。
4. 避坑排查：CDH 集成 Flink 最常见的五类翻车
4.1 ClassNotFoundException: org.apache.hadoop.yarn.client.api.YarnClient
现象：提交任务时报ClassNotFoundException，提示找不到 YARN 客户端类。
原因：Flink 集成包没有把 CDH 的 YARN jar 放进 classpath，或者env.yarn.conf.dir配置错误导致 YARN 配置没加载。
解决：确认flink-conf.yaml里env.yarn.conf.dir指向/etc/hadoop/conf，并且该目录下有yarn-site.xml。如果还不行，在提交命令前手动 export：
export HADOOP_CLASSPATH=`hadoop classpath`4.2 NoSuchMethodError: org.apache.hadoop.conf.Configuration.getPassword
现象：任务启动后抛NoSuchMethodError，方法签名对不上。
原因：classpath 里同时存在多个版本的 Hadoop jar，比如 Flink 自带的 shaded Hadoop 和 CDH 的 Hadoop 冲突。
解决：检查$FLINK_HOME/lib/下是否有flink-shaded-hadoop-*.jar，有就删掉。同时确认HADOOP_CLASSPATH里没有混入其他版本的 Hadoop。
4.3 YARN 容器被 Kill：Container killed by YARN for exceeding memory limits
现象：TaskManager 容器运行一段时间后被 YARN 杀掉，日志显示内存超限。
原因：taskmanager.memory.process.size设置过大，超过了 YARN 队列的单容器上限；或者 JVM 堆外内存没算进去。
解决：先查 YARN 的yarn.scheduler.maximum-allocation-mb，确保taskmanager.memory.process.size不超过这个值。Flink 1.13 的内存模型把 JVM Metaspace、JVM Overhead 都算在 process size 里，一般建议 process size 比实际堆内存大 20% 左右。
4.4 Checkpoint 失败：Could not find any factory for identifier 'hdfs'
现象：Checkpoint 一直失败，日志提示找不到 HDFS 文件系统工厂。
原因：Flink 的lib/目录下缺少flink-shaded-hadoop-3-uber或者对应的 HDFS 依赖。
解决：CDH 环境下，如果集成包已经正确编译，HDFS 依赖应该来自 CDH 的 classpath。检查HADOOP_CLASSPATH是否包含hadoop-hdfs-client相关 jar。如果用的是filesystemstate backend，确保state.checkpoints.dir的 HDFS 路径可写。
4.5 SQL Client 启动报 Hive 元数据连接超时
现象：SQL Client 启动时卡在Connecting to Hive Metastore，最终超时。
原因：Hive Metastore 地址配置错误，或者网络不通。
解决：检查/etc/hive/conf/hive-site.xml里的hive.metastore.uris是否正确。CDH 环境下 Hive Metastore 通常由 Cloudera Manager 管理，确认服务状态正常。如果 Metastore 开了 Kerberos，还需要配置security.kerberos.login.keytab和principal。
5. 进阶技巧：用 Flink SQL 打通 Hive 数仓的实时链路
前面把环境跑通之后，真正体现这套集成价值的地方是 Flink SQL 直接读写 Hive 表。CDH 6.3.2 的 Hive 2.1.1 支持 ACID 事务表，Flink 1.13.1 的 Hive 连接器对 Hive 2.1.x 的读写兼容性已经比较成熟，但有几个细节需要特别注意。
第一个是 Hive 表的流式读取。Flink 读 Hive 表默认是批模式，如果要持续监控 Hive 表的新分区，需要开启streaming-source.enable：
CREATE TABLE hive_orders ( order_id BIGINT, user_id BIGINT, amount DECIMAL(10,2), dt STRING ) PARTITIONED BY (dt) WITH ( 'connector' = 'hive', 'streaming-source.enable' = 'true', 'streaming-source.monitor-interval' = '5 min', 'streaming-source.consume-start-offset' = '2024-01-01' );streaming-source.monitor-interval控制扫描新分区的频率，太短会给 Hive Metastore 压力，太长则实时性差，5 分钟是个折中值。consume-start-offset指定从哪个分区开始消费，不设的话默认从最新分区开始。
第二个是 Hive 写入的小文件问题。Flink 写 Hive 表时，每个 checkpoint 周期会生成一批小文件，如果 checkpoint 间隔短、并发高，HDFS 上很快就是一堆小文件。解决办法是开启 Hive 表的 compaction，或者调整 Flink 的sink.partition-commit.policy.kind为metastore,success-file，配合 Hive 的hive.compactor服务定期合并。
第三个是 Kerberos 环境下的认证。CDH 集群如果开了 Kerberos，Flink 提交任务时需要指定 keytab：
$FLINK_HOME/bin/flink run-application \ -t yarn-application \ -Dsecurity.kerberos.login.keytab=/etc/security/keytabs/flink.keytab \ -Dsecurity.kerberos.login.principal=flink@EXAMPLE.COM \ -Dsecurity.kerberos.login.use-ticket-cache=false \ -c com.example.MyJob \ /path/to/my-job.jaruse-ticket-cache设为 false 是强制 Flink 用 keytab 重新认证，避免 ticket cache 过期导致任务失败。这个配置在长时间运行的流任务里尤其重要，我见过太多因为 ticket 过期导致任务半夜挂掉的案例。
最后说一个验证方法：任务提交后，别只看 YARN 的 Web UI 显示RUNNING就完事。去 Flink 的 Web UI 看 Checkpoint 页面，确认 checkpoint 持续成功；再去 HDFS 上看state.checkpoints.dir下有没有新的 checkpoint 目录生成；最后查 Hive 表的分区有没有新数据写入。这三步都过了，才算真正跑通。
从那以后我每次在 CDH 上部署 Flink，都强制走一遍「版本对齐表 → 依赖检查 → 小任务验证 → 全链路压测」的流程，再也不敢跳过任何一步。希望帮到你。
本文还有配套的精品资源，点击获取
