简介：本资源面向大数据开发与运维工程师，提供 Flink 1.13.1 在 CDH 6.3.2 集群上的完整部署包，解决在 YARN 环境下集成流处理框架时的版本匹配与分发难题。压缩包共 5 个文件，约 299.52MB，包含 Flink 核心库与 YARN 客户端 jar、Parcel 二进制发行包及其 SHA 校验文件、Parcel 元数据 json，分别用于程序开发、任务提交、CDH 分发安装与完整性验证。已有 3496 人学习下载，适合需要快速搭建实时计算环境的中高级用户。借助该资源，读者可省去手动编译与依赖整理，直接完成 Parcel 导入、激活与集群分发，并基于 YARN 提交和管理 Flink 作业，同时掌握版本兼容、资源调度与故障排查思路，为后续实时数据处理项目提供稳定基础。
1. 为什么 CDH 6.3.2 上装 Flink 1.13.1 总有人翻车
CDH 6.3.2 这套发行版里自带的是 Spark，Flink 得自己往里塞。我见过太多人在这上面栽跟头：有人直接把 Apache 官方二进制包解压到/opt就以为完事了，结果一提交任务就报NoClassDefFoundError；有人把 Flink 的 jar 往 HDFS 上一扔，配了个high-availability就重启，结果 JobManager 起不来，日志里全是Connection refused。问题不在 Flink 本身，在于 CDH 6.3.2 用的是 JDK 8、Hadoop 3.0.0、Zookeeper 3.4.5 这一套组合，而 Flink 1.13.1 默认编译时依赖的 Hadoop 版本和 CDH 打包的版本对不上。你要是拿官方flink-1.13.1-bin-hadoop2.7的包直接跑在 CDH 6.3.2 上，Hadoop 类库冲突几乎是必然的。这篇笔记就是把我自己在一套三节点 CDH 6.3.2 集群上装 Flink 1.13.1 的过程拆开，从选包、改配置、对接 YARN 到验证跑通，每一步都落到命令和参数上。适合正在 CDH 环境里补 Flink 组件的运维和开发，也适合想搞清楚 Flink on YARN 到底怎么配的人。
2. 装之前先把环境对齐：CDH 6.3.2 与 Flink 1.13.1 的版本账
2.1 先确认 CDH 6.3.2 自带的组件版本
在动手之前，我一般会先把集群现有版本摸一遍。CDH 6.3.2 对应的核心组件版本是固定的：Hadoop 3.0.0、Zookeeper 3.4.5、Hive 2.1.1、Spark 2.4.0。这些版本决定了 Flink 1.13.1 该选哪个二进制包。很多人忽略这一步，直接下flink-1.13.1-bin-scala_2.12.tgz，那个包默认绑的是 Hadoop 2.8.3 的类库，放到 CDH 6.3.2 上跑 YARN 模式，Hadoop 的Configuration类加载顺序会乱，表现就是NoSuchMethodError或者ClassNotFoundException。
确认命令很简单，在任意一个 CDH 节点上执行：
# 查看 Hadoop 版本 hadoop version | head -n 1 # 查看 Zookeeper 版本（CDH 自带） ls /opt/cloudera/parcels/CDH/lib/zookeeper/lib/ | grep zookeeper-.*\.jar # 查看 JDK 版本 java -version 2>&1 | head -n 1输出里 Hadoop 应该是Hadoop 3.0.0，JDK 是1.8.0_xxx。如果 JDK 是 11，那 Flink 1.13.1 虽然支持，但 CDH 6.3.2 的 Hive 和 Spark 在 JDK 11 下会有反射警告，我建议统一用 JDK 8，别给自己找事。
2.2 选对 Flink 1.13.1 的二进制包
Flink 1.13.1 官方提供了多个预编译包，区别在于捆绑的 Hadoop 和 Scala 版本。针对 CDH 6.3.2，我选的是flink-1.13.1-bin-scala_2.12.tgz，但不会直接用里面自带的 Hadoop 类库。原因是 CDH 6.3.2 的 Hadoop 是 3.0.0，而 Flink 1.13.1 官方包默认带的是 Hadoop 2.8.3 的客户端 jar。正确做法是：下载flink-1.13.1-bin-scala_2.12.tgz，然后手动把 CDH 的 Hadoop jar 包替换进去，或者干脆用flink-1.13.1-bin-hadoop3.tgz这个包——它捆绑的是 Hadoop 3.x 的类库，和 CDH 6.3.2 的 Hadoop 3.0.0 兼容性更好。
我实际用的是flink-1.13.1-bin-hadoop3.tgz，下载地址从 Apache 归档站拿，这里不贴外链，搜「flink-1.13.1-bin-hadoop3.tgz」就能找到。下载后校验一下：
# 下载后校验文件完整性 sha512sum flink-1.13.1-bin-hadoop3.tgz # 解压到 /opt 下 tar -zxvf flink-1.13.1-bin-hadoop3.tgz -C /opt/ mv /opt/flink-1.13.1 /opt/flink解压完先别急着配，看一眼/opt/flink/lib下面有没有flink-shaded-hadoop-3-uber-3.1.1.7.2.9.0-148-2.0.jar这类包。如果有，说明 Hadoop 3 的依赖已经打进去了，后面配 YARN 会省心很多。
2.3 环境变量与目录规划
Flink 在 CDH 集群上跑，我习惯把它放在/opt/flink，日志单独挂一个盘或者放到/data/flink/logs，避免把系统盘写满。环境变量在/etc/profile.d/flink.sh里配：
# /etc/profile.d/flink.sh export FLINK_HOME=/opt/flink export PATH=$PATH:$FLINK_HOME/bin # 让 Flink 能找到 CDH 的 Hadoop 配置 export HADOOP_CONF_DIR=/etc/hadoop/conf export HADOOP_CLASSPATH=`hadoop classpath`这里HADOOP_CONF_DIR必须指向 CDH 的 Hadoop 配置目录，通常是/etc/hadoop/conf。HADOOP_CLASSPATH用hadoop classpath命令动态生成，这样 Flink 提交到 YARN 时才能拿到正确的 core-site.xml 和 hdfs-site.xml。配完执行source /etc/profile.d/flink.sh，然后echo $HADOOP_CONF_DIR确认一下。
提示：如果集群开了 Kerberos，
HADOOP_CONF_DIR下还得有对应的 keytab 和 krb5.conf，Flink 提交任务时需要指定 principal 和 keytab 路径，这个后面在 YARN 配置里说。
3. 配置 Flink 1.13.1：从 flink-conf.yaml 到 YARN 对接
3.1 核心配置项：jobmanager、taskmanager 与内存
Flink 1.13.1 的配置文件在/opt/flink/conf/flink-conf.yaml。这个版本的内存模型已经统一成taskmanager.memory.process.size和jobmanager.memory.process.size，不再用老的taskmanager.heap.size。我一般按下面这样改：
# /opt/flink/conf/flink-conf.yaml # JobManager 进程总内存，按集群规模给 jobmanager.memory.process.size: 2048m # TaskManager 进程总内存，单节点 64G 内存给 8G 比较稳 taskmanager.memory.process.size: 8192m # 每个 TaskManager 的 slot 数，和 CPU 核数对齐 taskmanager.numberOfTaskSlots: 4 # 并行度默认值，按业务调整 parallelism.default: 4 # 检查点存储，CDH 环境一般写 HDFS 路径 state.backend: filesystem state.checkpoints.dir: hdfs:///flink/checkpoints state.savepoints.dir: hdfs:///flink/savepoints # 开启检查点，间隔 60 秒 execution.checkpointing.interval: 60000 # 高可用配置，依赖 Zookeeper high-availability: zookeeper high-availability.storageDir: hdfs:///flink/ha high-availability.zookeeper.quorum: zk-node1:2181,zk-node2:2181,zk-node3:2181 high-availability.zookeeper.path.root: /flink high-availability.cluster-id: /default这里有几个参数容易写错。high-availability.zookeeper.quorum要填 CDH 的 Zookeeper 地址，端口默认 2181，CDH 6.3.2 的 Zookeeper 一般没改端口。high-availability.storageDir必须是一个 HDFS 路径，Flink 会把 JobManager 的元数据写进去，这个路径需要提前创建：
# 在 HDFS 上创建 Flink 需要的目录 hdfs dfs -mkdir -p /flink/checkpoints hdfs dfs -mkdir -p /flink/savepoints hdfs dfs -mkdir -p /flink/ha hdfs dfs -chmod -R 755 /flinktaskmanager.memory.process.size这个值要算一下：进程总内存 = JVM 元空间 + JVM 开销 + 框架堆 + 框架堆外 + 任务堆 + 任务堆外 + 网络内存。Flink 1.13.1 会自动按比例切分，你只需要给总大小。我一般给 8G 到 16G，看单节点内存。如果给太小，任务一跑就报OutOfMemoryError: Direct buffer memory，那就是网络内存不够，得往上加总内存。
3.2 对接 YARN：让 Flink 跑在 CDH 的资源池里
CDH 6.3.2 的 YARN 是主力资源调度器，Flink 1.13.1 支持per-job和application两种模式。我推荐用application模式，一个 Flink 应用对应一个 YARN Application，JobManager 和 TaskManager 都在 YARN 容器里跑，资源隔离干净。
配置上，除了flink-conf.yaml，还要确认 YARN 的capacity-scheduler.xml里有足够的队列资源。Flink 提交时会用yarn.application.queue指定队列，默认是default。如果 CDH 里配了动态资源池，得确保这个队列存在且有资源。
提交命令长这样：
# 以 application 模式提交一个 Flink 作业到 YARN /opt/flink/bin/flink run-application \ -t yarn-application \ -Djobmanager.memory.process.size=2048m \ -Dtaskmanager.memory.process.size=8192m \ -Dtaskmanager.numberOfTaskSlots=4 \ -Dyarn.application.name=flink-demo \ -Dyarn.application.queue=root.default \ -c com.example.MyFlinkJob \ /opt/flink/examples/streaming/MyFlinkJob.jar参数说明：-t yarn-application指定部署模式；-D后面跟的是动态配置，会覆盖flink-conf.yaml里的同名项；-c指定主类；最后是 jar 包路径。如果集群开了 Kerberos，还要加：
-Dyarn.security.kerberos.keytab=/etc/security/keytabs/flink.keytab \ -Dyarn.security.kerberos.principal=flink@EXAMPLE.COM提交后，用yarn application -list能看到一个名为flink-demo的应用。如果状态一直是ACCEPTED不动，多半是队列资源不够，去 YARN 的 ResourceManager Web UI 看队列容量。
3.3 与 HDFS、Hive 的集成配置
Flink 1.13.1 读写 HDFS 不需要额外配，只要HADOOP_CONF_DIR指对了，fs.defaultFS会自动从core-site.xml里读。但如果要用 Flink SQL 读 Hive 表，就得把 Hive 的依赖加进来。CDH 6.3.2 的 Hive 是 2.1.1，Flink 1.13.1 对应的 Hive connector 是flink-connector-hive_2.12-1.13.1.jar，还要加 Hive 的hive-exec和hive-metastore包。
我一般这么做：
# 把 Hive connector 和依赖拷到 Flink 的 lib 目录 cp /opt/flink/opt/flink-connector-hive_2.12-1.13.1.jar /opt/flink/lib/ cp /opt/cloudera/parcels/CDH/lib/hive/lib/hive-exec-2.1.1.jar /opt/flink/lib/ cp /opt/cloudera/parcels/CDH/lib/hive/lib/hive-metastore-2.1.1.jar /opt/flink/lib/然后在 SQL Client 里配置 Hive catalog：
-- 在 Flink SQL Client 中创建 Hive Catalog CREATE CATALOG hive_catalog WITH ( 'type' = 'hive', 'default-database' = 'default', 'hive-conf-dir' = '/etc/hive/conf' ); USE CATALOG hive_catalog; SHOW DATABASES;hive-conf-dir指向 CDH 的 Hive 配置目录，里面要有hive-site.xml。如果连不上 Hive Metastore，先检查hive-site.xml里的hive.metastore.uris是否可达，以及 Kerberos 票据是否有效。
4. 避坑与排查：CDH 6.3.2 上跑 Flink 1.13.1 的五个血泪坑
4.1 坑一：ClassNotFoundException: org.apache.hadoop.conf.Configuration
现象：提交任务到 YARN 后，JobManager 日志里报java.lang.ClassNotFoundException: org.apache.hadoop.conf.Configuration，任务直接失败。
原因：Flink 的lib目录下没有 Hadoop 的类库，或者HADOOP_CLASSPATH没生效。CDH 6.3.2 的 Hadoop jar 在/opt/cloudera/parcels/CDH/lib/hadoop/client/下，Flink 默认不会去那里找。
解决：在flink-conf.yaml里加一行classloader.parent-first-patterns.additional: org.apache.hadoop，或者更直接的办法是把 CDH 的 Hadoop jar 软链到 Flink 的 lib 下：
# 把 CDH Hadoop 客户端 jar 链到 Flink lib ln -s /opt/cloudera/parcels/CDH/lib/hadoop/client/*.jar /opt/flink/lib/我一般用后者，简单粗暴，但注意别把冲突的 jar 也链进去，比如 Hadoop 自带的guava版本和 Flink 的不一致，会引发NoSuchMethodError。如果出现 guava 冲突，把 Flink lib 下的 guava 删掉，用 CDH 的。
4.2 坑二：Zookeeper 连接超时导致 HA 切换失败
现象：JobManager 挂掉后，备用的 JobManager 起不来，日志里刷Connection timed out连 Zookeeper。
原因：high-availability.zookeeper.quorum配错了，或者 CDH 的 Zookeeper 开了 ACL，Flink 没权限创建 znode。
解决：先用zkCli.sh连一下 Zookeeper，确认地址和端口对：
# 测试 Zookeeper 连通性 /opt/cloudera/parcels/CDH/lib/zookeeper/bin/zkCli.sh -server zk-node1:2181连上后执行ls /flink，看能不能创建节点。如果报NoAuth，说明 Zookeeper 配了 ACL，需要在flink-conf.yaml里加high-availability.zookeeper.client.acl: open或者对应的认证信息。CDH 默认的 Zookeeper 一般没开 ACL，但如果你在 CDH 里启用了 Zookeeper 的 SASL，那就得配zookeeper.sasl.client: true和 jaas 文件。
4.3 坑三：TaskManager 内存超限被 YARN 杀掉
现象：任务跑着跑着，TaskManager 容器被 YARN 杀掉，日志显示Container killed by YARN for exceeding memory limits。
原因：taskmanager.memory.process.size设小了，或者 Flink 的堆外内存没算进去。YARN 监控的是容器物理内存，Flink 的 JVM 元空间、直接内存都算在里面。
解决：把taskmanager.memory.process.size调大，同时确认taskmanager.memory.jvm-metaspace.size和taskmanager.memory.jvm-overhead.fraction没被改乱。我一般会留 20% 的余量，比如 YARN 容器给 10G，Flink 的 process.size 就设 8G。另外，如果用了 RocksDB 状态后端，state.backend.rocksdb.memory.managed: true要打开，让 Flink 自己管内存。
4.4 坑四：HDFS 权限不足导致检查点失败
现象：检查点一直失败，日志报Permission denied: user=flink, access=WRITE, inode="/flink/checkpoints"。
原因：Flink 提交任务时用的用户和 HDFS 目录的 owner 不一致。CDH 里通常用hdfs或yarn用户跑，但 Flink 可能以提交者身份写 HDFS。
解决：把 HDFS 目录的权限放开，或者用HADOOP_USER_NAME指定用户：
# 提交前指定 HDFS 用户 export HADOOP_USER_NAME=hdfs或者在 HDFS 上把/flink目录改成 777（测试环境可以，生产环境建议用 ACL）：
hdfs dfs -chmod -R 777 /flink4.5 坑五：Scala 版本不匹配导致作业提交失败
现象：提交一个用 Scala 写的 Flink 作业，报NoSuchMethodError: scala.Predef$.$conforms()。
原因：Flink 1.13.1 的包是 Scala 2.12 编译的，但你的作业 jar 是用 Scala 2.11 编译的。CDH 6.3.2 自带的 Spark 是 Scala 2.11 的，很多人顺手就用 2.11 编 Flink 作业，结果版本对不上。
解决：统一用 Scala 2.12 编译 Flink 作业。在 Maven 的pom.xml里把scala.binary.version设成 2.12，并且确保flink-streaming-scala_2.12依赖版本是 1.13.1。如果作业里还依赖了 Spark 的包，那得小心，Spark 2.4.0 是 Scala 2.11 的，混用会出问题，最好把 Spark 依赖排除掉。
5. 装完怎么验证：从 WordCount 到 YARN 日志的完整检查链
装完 Flink 1.13.1 后，别急着上生产作业，先跑一个自带的 WordCount 例子，把整条链路走通。验证分三步：本地模式、YARN 模式、检查点恢复。
5.1 本地模式快速验证
先用本地模式确认 Flink 本身能跑起来：
# 启动本地 Flink 集群（前台运行，方便看日志） /opt/flink/bin/start-cluster.sh # 提交自带的 WordCount 例子 /opt/flink/bin/flink run /opt/flink/examples/streaming/WordCount.jar # 查看 JobManager Web UI # 浏览器访问 http://<jobmanager-host>:8081如果 8081 端口能打开，看到 WordCount 任务状态是FINISHED，说明 Flink 基础环境没问题。如果端口打不开，检查jobmanager.rpc.address和rest.address配置，CDH 环境里主机名解析要确保正确，/etc/hosts里每个节点都要能互相解析。
5.2 YARN 模式验证与日志定位
本地模式过了之后，切到 YARN 模式：
# 提交 WordCount 到 YARN /opt/flink/bin/flink run-application \ -t yarn-application \ -Dyarn.application.name=wordcount-test \ /opt/flink/examples/streaming/WordCount.jar # 查看 YARN 应用列表 yarn application -list | grep wordcount-test # 查看 Flink 作业的 YARN 日志 yarn logs -applicationId <application_id> | grep -i "error\|exception"YARN 模式跑通的标准是：yarn application -list里状态为RUNNING或FINISHED，并且 Flink Web UI 能通过 YARN 的 ApplicationMaster 代理访问。如果任务卡在ACCEPTED，去 ResourceManager UI 看队列资源；如果卡在RUNNING但没输出，去yarn logs里搜Checkpoint和TaskManager关键字。
5.3 检查点与恢复验证
最后一步，验证检查点能不能正常写 HDFS 并恢复：
# 提交一个带检查点的作业，手动触发 savepoint /opt/flink/bin/flink run-application \ -t yarn-application \ -Dexecution.checkpointing.interval=10000 \ -Dstate.checkpoints.dir=hdfs:///flink/checkpoints \ -Dyarn.application.name=checkpoint-test \ /opt/flink/examples/streaming/WordCount.jar # 查看 HDFS 上的检查点目录 hdfs dfs -ls /flink/checkpoints/<job_id>/如果 HDFS 上能看到chk-1、chk-2这样的目录，说明检查点正常。然后手动触发一次 savepoint：
# 触发 savepoint /opt/flink/bin/flink savepoint <job_id> hdfs:///flink/savepoints # 从 savepoint 恢复 /opt/flink/bin/flink run-application \ -t yarn-application \ -s hdfs:///flink/savepoints/savepoint-<id> \ /opt/flink/examples/streaming/WordCount.jar恢复成功的话，任务会从上次的状态继续跑，不会从头开始。这一步过了，整套 Flink on CDH 6.3.2 的环境就算真正落地了。
从那以后，我每次在 CDH 上装 Flink，都会先把hadoop classpath打出来，对着 jar 列表检查一遍冲突，再跑 WordCount 和检查点恢复。这套流程走完，后面上生产作业心里才有底。希望帮到你。
本文还有配套的精品资源，点击获取
