简介本资源面向大数据开发与运维工程师提供 Flink 1.13.1 在 CDH 6.3.2 集群上的完整部署包解决在 YARN 环境下集成流处理框架时的版本适配与分发难题。压缩包共 5 个文件约 299.52MB包含两个 jar 核心库与 YARN 客户端、一个 Parcel 二进制发行包及其 SHA 校验文件以及描述版本依赖的 manifest.json 元数据覆盖从 Parcel 导入、激活到作业提交所需的全部组件。已有 3496 人学习下载适合需要快速搭建实时计算环境的中高级用户。借助该包可省去逐一下载依赖的繁琐直接完成 CDH 集群上的 Flink 服务配置与任务调度并对照校验文件确保分发安全为后续实时数据处理与故障排查提供稳定基础。1. 为什么 CDH 6.3.2 上装 Flink 1.13.1 值得单独写一篇在 CDH 6.3.2 集群上装 Flink 1.13.1看起来只是“解压、配环境变量、提交任务”三步但真到生产里翻车点往往不在 Flink 本身而在 CDH 的 JDK 版本、Hadoop 依赖冲突和 YARN 队列权限上。我见过最典型的一次是某开发者在 CDH 6.3.2 的节点上直接跑 Flink 1.13.1 的默认发行包结果flink run一提交就报NoClassDefFoundError: org/apache/hadoop/yarn/client/api/YarnClient排查半天才发现是 Flink 自带的 Hadoop 依赖和 CDH 集群里的 Hadoop 版本对不上。CDH 6.3.2 自带的是 Hadoop 3.0.0 和 JDK 1.8而 Flink 1.13.1 官方预编译包默认绑的是 Hadoop 2.7 的依赖这就是版本坑的根源。这套组合适合已经在用 CDH 6.3.2 做离线批处理、又想引入 Flink 做实时计算的团队也适合需要把 Flink 跑在 YARN 上、复用现有集群资源的场景。装不好轻则任务提交失败重则和 Hive、Spark 抢资源导致整个集群不稳定。2. 装之前先把环境对齐CDH 6.3.2 与 Flink 1.13.1 的版本账2.1 确认 CDH 6.3.2 的 JDK 和 Hadoop 版本CDH 6.3.2 官方支持的 JDK 是 Oracle JDK 1.8Hadoop 版本是 3.0.0。在动手装 Flink 之前先在任意一个 CDH 节点上执行下面两条命令确认# 查看当前 JDK 版本必须是 1.8 java -version # 查看 Hadoop 版本CDH 6.3.2 对应的是 3.0.0 hadoop versionjava -version输出里如果出现1.8.0_xxx就对了如果是 JDK 11 或 17Flink 1.13.1 虽然理论上支持但和 CDH 6.3.2 的 Hive、Spark 组件混用时容易出现序列化兼容问题我一般会强制统一到 JDK 1.8。hadoop version的输出里会明确写Hadoop 3.0.0这个版本号决定了后面下载 Flink 发行包时要选哪个 Hadoop 绑定版本。2.2 选对 Flink 1.13.1 的发行包Flink 1.13.1 官方提供了多个预编译版本区别在于绑定的 Hadoop 版本不同。在 CDH 6.3.2 环境下不能直接用默认的flink-1.13.1-bin-scala_2.11.tgz因为它绑的是 Hadoop 2.7。正确的做法是下载带 Hadoop 3.0 绑定的包或者下载不带 Hadoop 的纯净包再手动指定 Hadoop 依赖。常见做法是去 Flink 官网下载flink-1.13.1-bin-scala_2.11.tgz后再额外下载flink-shaded-hadoop-3-uber-3.1.1.7.2.9.0-148-2.0.jar放到lib目录下。但更省事的做法是直接找带hadoop3标识的发行包。如果内网无法直接下载可以从其他节点拷贝已经验证过的包。提示Scala 版本也要和集群里已有的 Spark、Hive 对齐。CDH 6.3.2 默认带的是 Scala 2.11所以选scala_2.11的 Flink 包不要选 2.12。2.3 规划安装目录和用户权限Flink 在 CDH 集群上通常以flink用户运行不要用 root 直接跑任务。先在所有要跑 Flink 的节点上创建用户和目录# 创建 flink 用户组和用户 groupadd flink useradd -g flink -m -s /bin/bash flink # 创建安装目录和数据目录 mkdir -p /opt/flink-1.13.1 mkdir -p /data/flink/checkpoints mkdir -p /data/flink/savepoints # 把目录权限给 flink 用户 chown -R flink:flink /opt/flink-1.13.1 chown -R flink:flink /data/flink/opt/flink-1.13.1是安装目录后面所有配置文件和 jar 包都放这里。/data/flink/checkpoints和/data/flink/savepoints分别用于 Flink 的检查点和保存点这两个目录必须提前建好并授权否则任务跑起来后写检查点会报权限拒绝。我一般会把这两个目录挂到独立的磁盘上避免和系统盘抢 IO。3. 安装 Flink 1.13.1从解压到 YARN 模式配置3.1 解压安装包并设置环境变量把下载好的 Flink 1.13.1 安装包上传到/opt目录然后切换到flink用户解压# 切换到 flink 用户 su - flink # 解压安装包到指定目录 tar -zxvf /opt/flink-1.13.1-bin-scala_2.11.tgz -C /opt/ # 进入解压后的目录确认结构 cd /opt/flink-1.13.1 ls -l解压后应该能看到bin、conf、lib、examples等目录。如果解压出来的目录名不是flink-1.13.1手动改一下或者建软链接。接下来配置环境变量编辑flink用户的.bashrc# 编辑 flink 用户的环境变量 vi ~/.bashrc # 在文件末尾追加以下内容 export FLINK_HOME/opt/flink-1.13.1 export PATH$PATH:$FLINK_HOME/bin export HADOOP_CONF_DIR/etc/hadoop/conf export HADOOP_CLASSPATHhadoop classpathFLINK_HOME指向安装目录PATH里加上bin目录后就能直接敲flink命令。HADOOP_CONF_DIR必须指向 CDH 的 Hadoop 配置目录CDH 6.3.2 默认是/etc/hadoop/conf这个目录里有core-site.xml、hdfs-site.xml、yarn-site.xmlFlink 提交 YARN 任务时要读这些配置。HADOOP_CLASSPATH通过hadoop classpath命令动态生成把 Hadoop 的所有依赖 jar 都塞进 Flink 的类路径里这一步是解决NoClassDefFoundError的关键。执行source ~/.bashrc让环境变量生效然后验证# 验证 Flink 命令是否可用 flink --version # 验证 Hadoop 类路径是否包含必要 jar echo $HADOOP_CLASSPATH | tr : \n | grep yarnflink --version应该输出Version: 1.13.1。第二条命令如果能看到yarn相关的 jar 路径说明HADOOP_CLASSPATH配置正确。3.2 配置 flink-conf.yaml 对接 YARNFlink 1.13.1 在 YARN 上跑核心配置都在conf/flink-conf.yaml里。下面是我在 CDH 6.3.2 上验证过的一套最小可用配置# conf/flink-conf.yaml # JobManager 和 TaskManager 的通信地址YARN 模式下由集群自动分配 jobmanager.rpc.address: localhost jobmanager.rpc.port: 6123 # JobManager 内存根据集群规模调整我一般给 2g jobmanager.memory.process.size: 2048m # TaskManager 内存CDH 节点内存充足的话给 4g taskmanager.memory.process.size: 4096m # 每个 TaskManager 的 slot 数建议等于 CPU 核数 taskmanager.numberOfTaskSlots: 4 # 并行度默认值YARN 模式下会被提交参数覆盖 parallelism.default: 4 # 检查点存储目录指向之前创建的 HDFS 路径 state.backend: filesystem state.checkpoints.dir: hdfs:///data/flink/checkpoints state.savepoints.dir: hdfs:///data/flink/savepoints # 开启检查点间隔 60 秒 execution.checkpointing.interval: 60000 # YARN 相关的配置 yarn.application-attempts: 3 yarn.per-job-cluster.include-user-jar: truejobmanager.memory.process.size和taskmanager.memory.process.size是 Flink 1.13 引入的新内存模型参数直接控制进程总内存比老版本的jobmanager.heap.size更直观。taskmanager.numberOfTaskSlots设成 4 是因为 CDH 节点通常有 8 核以上留一半给系统和其他组件。state.checkpoints.dir和state.savepoints.dir必须用 HDFS 路径不能用本地路径否则 YARN 容器重启后检查点就丢了。yarn.application-attempts: 3表示 YARN 应用失败后最多重试 3 次生产环境建议保留。注意state.checkpoints.dir里的 HDFS 路径要提前在 HDFS 上创建并授权给flink用户否则任务启动时会报Permission denied。3.3 配置 YARN 队列和日志目录CDH 6.3.2 的 YARN 通常有多个队列Flink 任务默认提交到default队列。如果集群有专门的实时队列需要在提交时指定或者改flink-conf.yaml# 指定默认提交队列根据实际队列名修改 yarn.application.queue: root.realtime # 日志目录指向 HDFS 上的路径 yarn.ship-files: /opt/flink-1.13.1yarn.application.queue的值要和你 CDH 集群里实际的队列名一致可以通过yarn queue -status命令查看。yarn.ship-files指定要分发到 YARN 容器里的文件目录一般就是 Flink 安装目录。如果队列名写错提交任务时会直接报Queue not found。另外Flink 在 YARN 上跑的时候日志默认写到容器本地目录容器销毁后日志就没了。我一般会配置日志聚合到 HDFS# 在 yarn-site.xml 里确认日志聚合已开启 # 这个配置在 CDH 的 YARN 配置里通常已经开启 grep -A2 yarn.log-aggregation-enable /etc/hadoop/conf/yarn-site.xml如果输出true说明日志聚合已开启Flink 容器日志会自动上传到 HDFS通过yarn logs -applicationId appId就能查看。4. 避坑与排查CDH 6.3.2 上跑 Flink 1.13.1 的五个血泪坑4.1 坑一NoClassDefFoundError 找不到 YARN 类现象执行flink run -m yarn-cluster提交任务时报NoClassDefFoundError: org/apache/hadoop/yarn/client/api/YarnClient。原因Flink 1.13.1 默认发行包绑的是 Hadoop 2.7 的依赖而 CDH 6.3.2 用的是 Hadoop 3.0.0类路径里找不到 Hadoop 3 的 YARN 客户端类。解决确认HADOOP_CLASSPATH环境变量已正确设置并且flink-conf.yaml里没有覆盖掉它。如果还不行把 CDH 的 Hadoop jar 包手动拷到 Flink 的lib目录# 从 CDH 的 Hadoop 目录拷贝 YARN 相关 jar 到 Flink lib cp /opt/cloudera/parcels/CDH/lib/hadoop-yarn/*.jar /opt/flink-1.13.1/lib/ cp /opt/cloudera/parcels/CDH/lib/hadoop/*.jar /opt/flink-1.13.1/lib/4.2 坑二检查点写 HDFS 报权限拒绝现象任务启动后日志里反复报Permission denied: userflink, accessWRITE, inode/data/flink/checkpoints。原因HDFS 上的/data/flink/checkpoints目录属主是hdfs或rootflink用户没有写权限。解决用hdfs用户登录把目录权限改成flink可写# 切换 hdfs 用户 su - hdfs # 修改目录属主和权限 hdfs dfs -chown -R flink:flink /data/flink/checkpoints hdfs dfs -chown -R flink:flink /data/flink/savepoints hdfs dfs -chmod -R 755 /data/flink4.3 坑三TaskManager 内存超限被 YARN 杀掉现象任务跑几分钟后YARN 报Container killed by YARN for exceeding memory limitsTaskManager 进程消失。原因taskmanager.memory.process.size设得太大超过了 YARN 容器内存上限或者没算上 JVM 元空间和堆外内存。解决把taskmanager.memory.process.size调小同时确认 YARN 的yarn.nodemanager.resource.memory-mb足够大。我一般会留 20% 的余量比如 YARN 单容器上限是 8gFlink 的 TaskManager 就设 6g。4.4 坑四Scala 版本不匹配导致序列化失败现象任务提交后报java.lang.NoSuchMethodError: scala.Predef$.refArrayOps或类似的 Scala 版本错误。原因Flink 包用的是 Scala 2.12而 CDH 6.3.2 集群里的 Spark、Hive 用的是 Scala 2.11类路径冲突。解决换用flink-1.13.1-bin-scala_2.11.tgz重新解压安装确保 Scala 版本和集群其他组件一致。4.5 坑五YARN 队列资源不足导致任务一直挂起现象任务提交后一直处于ACCEPTED状态不进入RUNNING。原因指定的 YARN 队列资源已满或者队列的maximum-am-resource-percent限制导致没有足够的资源启动 ApplicationMaster。解决先用yarn application -list查看队列里正在跑的任务确认资源占用情况。如果队列确实满了要么等资源释放要么换队列提交# 提交时指定其他队列 flink run -m yarn-cluster -yqu root.offline -c com.example.MyJob myjob.jar-yqu参数后面跟队列名临时覆盖flink-conf.yaml里的默认队列配置。5. 装完怎么验证从 WordCount 到 YARN 日志的完整检查链装完 Flink 1.13.1 后不要急着上生产任务先用官方自带的 WordCount 例子跑一遍确认整条链路是通的。验证分三步本地模式验证、YARN 模式验证、日志和 UI 检查。第一步本地模式跑 WordCount确认 Flink 本身没问题# 准备测试数据 echo -e flink cdh\nflink yarn\ncdh hadoop /tmp/test.txt # 本地模式运行 WordCount flink run -c org.apache.flink.examples.java.wordcount.WordCount \ /opt/flink-1.13.1/examples/batch/WordCount.jar \ --input /tmp/test.txt --output /tmp/result.txt # 查看结果 cat /tmp/result.txt如果输出里能看到flink、cdh、yarn等单词的计数说明 Flink 本地运行正常。这一步排除了 JDK 和 Scala 版本的问题。第二步YARN 模式提交同一个任务验证和 CDH 集群的对接# 提交到 YARN 集群 flink run -m yarn-cluster \ -yqu root.default \ -c org.apache.flink.examples.java.wordcount.WordCount \ /opt/flink-1.13.1/examples/batch/WordCount.jar \ --input hdfs:///tmp/test.txt --output hdfs:///tmp/result_yarn.txt提交后终端会输出一个Application ID类似application_1234567890_0001。记下这个 ID后面查日志和 UI 都要用。如果提交成功终端会显示Job has been submitted with JobID xxxxx。第三步检查 YARN 应用状态和 Flink UI# 查看 YARN 应用状态 yarn application -status application_1234567890_0001 # 查看 Flink 任务日志 yarn logs -applicationId application_1234567890_0001 | grep -A5 WordCountyarn application -status的输出里State应该是FINISHEDFinal-State是SUCCEEDED。如果状态是FAILED用yarn logs命令拉取完整日志重点看TaskManager和JobManager的报错信息。Flink 1.13.1 在 YARN 上的 UI 默认不对外开放需要通过 YARN 的代理地址访问通常是http://ResourceManager-IP:8088/proxy/application_1234567890_0001/。最后确认 HDFS 上的输出结果# 查看 YARN 模式下的输出 hdfs dfs -cat /tmp/result_yarn.txt如果能看到和本地模式一致的单词计数说明 Flink 1.13.1 在 CDH 6.3.2 上的安装和 YARN 对接全部跑通。从那以后我每次装完 Flink都会强制走一遍「本地 WordCount → YARN WordCount → 查 YARN 日志 → 查 HDFS 输出」这条检查链少一步都可能在上生产时才发现问题。希望帮到你。本文还有配套的精品资源点击获取
