简介本资源面向大数据开发与运维工程师提供 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 这套发行版里自带的是 SparkFlink 得自己往里塞。我见过太多人在这上面栽跟头有人直接把 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 21 | head -n 1输出里 Hadoop 应该是Hadoop 3.0.0JDK 是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_CLASSPATHhadoop 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确认一下。提示如果集群开了 KerberosHADOOP_CONF_DIR下还得有对应的 keytab 和 krb5.confFlink 提交任务时需要指定 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 地址端口默认 2181CDH 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 ApplicationJobManager 和 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.size2048m \ -Dtaskmanager.memory.process.size8192m \ -Dtaskmanager.numberOfTaskSlots4 \ -Dyarn.application.nameflink-demo \ -Dyarn.application.queueroot.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.principalflinkEXAMPLE.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.1Flink 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 开了 ACLFlink 没权限创建 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 容器给 10GFlink 的 process.size 就设 8G。另外如果用了 RocksDB 状态后端state.backend.rocksdb.memory.managed: true要打开让 Flink 自己管内存。4.4 坑四HDFS 权限不足导致检查点失败现象检查点一直失败日志报Permission denied: userflink, accessWRITE, inode/flink/checkpoints。原因Flink 提交任务时用的用户和 HDFS 目录的 owner 不一致。CDH 里通常用hdfs或yarn用户跑但 Flink 可能以提交者身份写 HDFS。解决把 HDFS 目录的权限放开或者用HADOOP_USER_NAME指定用户# 提交前指定 HDFS 用户 export HADOOP_USER_NAMEhdfs或者在 HDFS 上把/flink目录改成 777测试环境可以生产环境建议用 ACLhdfs 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.namewordcount-test \ /opt/flink/examples/streaming/WordCount.jar # 查看 YARN 应用列表 yarn application -list | grep wordcount-test # 查看 Flink 作业的 YARN 日志 yarn logs -applicationId application_id | grep -i error\|exceptionYARN 模式跑通的标准是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.interval10000 \ -Dstate.checkpoints.dirhdfs:///flink/checkpoints \ -Dyarn.application.namecheckpoint-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 和检查点恢复。这套流程走完后面上生产作业心里才有底。希望帮到你。本文还有配套的精品资源点击获取