简介面向在 CDH 6.3.2 集群上部署 Flink 1.13.1 的大数据运维与开发人员这份资源提供一套可用于 Parcel 分发和 YARN 提交的部署文件集合重点解决离线环境组件不全、版本匹配与依赖缺失等问题。压缩包共 5 个文件约 299.52MB主要由两个 jar、一个 parcel、一个 sha 校验文件和一个 json 元数据组成jar 覆盖 YARN 客户端与核心运行库parcel 承载 Scala 2.11、el7 环境的二进制发行版sha 与 json 则分别用于完整性校验和分发元信息。目前已有 3498 人学习。对于需要将 Flink 接入现有 Hadoop 生态的团队这套文件能帮助快速完成组件准备、版本核对与集群侧分发减少因包不完整或校验失败导致的反复排查其文件组合也便于在离线集群中核对依赖关系、统一部署口径并为后续作业提交提供一致的运行环境适合作为该平台部署的基础物料。1. 版本先对齐CDH 6.3.2 上跑 flink 1.13.1 的前置判断上周给一套 CDH 6.3.2 的集群加流计算层运维图省事从归档站直接拖了flink-1.13.1-bin-scala_2.12.tgz下来解压作业一提交就死在NoSuchMethodError: scala.Predef$.refArrayOps日志翻半天没有任何有用信息。这不是 Flink 坏了是 Scala 版本对不上——CDH 6.3.2 整条生态线是 Scala 2.11flink 1.13.1 的二进制包分 2.11 和 2.12 两套装错了连 Hello World 都跑不起来。更隐蔽的是 Hadoop从 1.11 起 Flink 的发行包不再内置 Hadoop 依赖CDH 6.3.2 又自带 Hadoop 3.0.0套用网上那些“解压即用”的老教程大概率卡在NoClassDefFoundError: org/apache/hadoop/fs/FileSystem。这篇按我实际落地的顺序走一遍选包、配环境、写flink-conf.yaml、提交 YARN、出事怎么救。2. 装前备料JDK、Scala 2.11 与 CDH 客户端的路径对齐装 Flink 最容易被跳过的一步就是“对版本”很多人上来就解压出问题再回头查来回折腾两小时。这一节把三件事定死JDK 用哪个、Scala 包怎么选、Hadoop 客户端从哪里来。这三样定完后面的配置基本是填空。2.1 一张表定死四个组件的版本先把 CDH 6.3.2 自带的版本和 flink 1.13.1 的要求摆在一起看冲突点一眼就能看出来组件CDH 6.3.2 自带flink 1.13.1 的要求落地结论JDK1.8.0 系列Java 8 主线支持Java 11 属实验态直接用集群自带 8别单独升级Scala2.11.x发行包分 scala_2.11 / scala_2.12必须选-bin-scala_2.11那个包Hadoop3.0.0二进制包不再内置 Hadoop 依赖靠HADOOP_CLASSPATH兜Hive2.1.1connector 官方覆盖 1.2.1 / 2.3.6 / 3.1.22.1.1 属于要绕的版本ZooKeeper3.4.53.4 及以上可用直接复用集群的不再单独装状态后端—RocksDB 依赖 glibc 2.17CentOS 7 基线满足不用额外装表里最后一行值得多说一句RocksDB 是 native 库libflink-runtime里那份.so对 glibc 版本有要求CentOS 6 的老机器上会抛UnsatisfiedLinkError这种机器别硬上换节点。至于 HiveCDH 6.3.2 的 2.1.1 不在官方列表里后面第 5 章会单独讲怎么处理。2.2 上传、解压与运行用户不要用 root 跑 Flink 进程CDH 集群里一般已经有hadoop这个服务账号直接复用最省事HDFS 权限也好理。包传到一台能连 HDFS 和 YARN 的边缘节点上解压到一个固定路径# 上传后解压到 /opt目录名统一改成 flink方便后面写环境变量 tar -zxvf flink-1.13.1-bin-scala_2.11.tgz -C /opt mv /opt/flink-1.13.1 /opt/flink # 属主给集群统一的服务账号避免 HDFS 写 checkpoint 时 Permission denied chown -R hadoop:hadoop /opt/flink chmod 755 /opt/flink # 确认包里的 Scala 版本没拿错 ls /opt/flink/lib | grep flink-distflink-dist_2.11-1.13.1.jar这个文件名里的2.11就是最后一道校验如果这里出现的是2.12不用往下走了重新下包。解压路径建议所有节点保持一致/opt/flink这种固定路径能省掉后面一堆路径写错的排查。2.3 环境变量与 HADOOP_CLASSPATH 的两种给法flink 1.13.1 的发行包不带 Hadoop所以要主动把 Hadoop 的类喂给它。业内有两种做法我一般优先选第一种因为不引入第二份 Hadoop 类冲突面小。第一种是让 Flink 用集群自己的 Hadoop 客户端cat /etc/profile.d/flink.sh EOF # Flink 主目录 export FLINK_HOME/opt/flink # CDH parcel 的解压目录因安装方式而异先探测再赋值避免写死 export CDH_HOME$(ls -d /opt/*/parcels/CDH 2/dev/null | head -1) # Hadoop 客户端与配置目录 export HADOOP_HOME$CDH_HOME/lib/hadoop export HADOOP_CONF_DIR/etc/hadoop/conf export YARN_CONF_DIR/etc/hadoop/conf export HBASE_CONF_DIR/etc/hbase/conf # 关键把 hadoop classpath 交给 Flink缺这行就是 NoClassDefFoundError export HADOOP_CLASSPATH$(hadoop classpath) export PATH$PATH:$FLINK_HOME/bin EOF source /etc/profile.d/flink.sh # 自检这两条都得有输出 echo $HADOOP_CONF_DIR hadoop classpath | head -c 200HADOOP_CLASSPATH的本质是把hadoop classpath的输出一长串 jar 路径用冒号分隔挂到 Flink 进程的 classpath 上。HADOOP_CONF_DIR则决定 Flink 能不能读到core-site.xml、hdfs-site.xml、yarn-site.xml提交到 YARN 时这些配置文件会被打进容器缺了就是连不上 ResourceManager。hadoop classpath命令本身依赖 CDH 的 alternatives 链接如果提示 command not found先用绝对路径$CDH_HOME/bin/hadoop classpath试一次能用就说明是 PATH 问题。第二种做法是往lib里塞一个 shaded 的 Hadoop uber jar好处是HADOOP_CLASSPATH不用管坏处是这份 jar 的 Hadoop 版本和 CDH 的 3.0.0 不完全一致某些 HDFS 新特性会踩空。两种方式不要同时用双份 Hadoop 类在 classpath 上打架报错信息会非常迷惑——这是血泪经验我见过一次ClassCastException: org.apache.hadoop.hdfs.DistributedFileSystem cannot be cast to ...查了一整天才定位到是两边都有。注意HADOOP_CLASSPATH内容很长某些场景会顶到命令行长度上限如果启动时报Argument list too long改用env.java.classpath或者只保留HADOOP_CONF_DIR把 Hadoop jar 交给 YARN 容器自带的 classpath。3. 配置 flink-conf.yaml内存、状态后端与 HA 三段写全conf/flink-conf.yaml是这套环境的黑匣子跑不起来十有八九能从这儿找到线索。flink 1.13.1 用的是 process 三级内存模型和 1.9 之前那套taskmanager.heap.size完全不是一回事照抄老配置会直接被忽略甚至报错。下面按基础、内存、状态与 HA 三段拆开写。3.1 基础项与内存模型的四个数先把通信和并行度定下来# ---- 基础通信 ---- jobmanager.rpc.address: node01 jobmanager.rpc.port: 6123 # 绑定 0.0.0.0 是为了让外部能访问 8081 的 Web UI jobmanager.bind-host: 0.0.0.0 taskmanager.bind-host: 0.0.0.0 # ---- 并行度 ---- parallelism.default: 4 taskmanager.numberOfTaskSlots: 4jobmanager.rpc.address写主机名还是 IP取决于集群/etc/hosts是否统一我一般写主机名方便后面换节点。taskmanager.host这一项在 standalone 多机部署时必须逐台改成自己否则 TaskManager 会拿127.0.0.1去注册JobManager 永远看不到 slot但在 YARN 模式下不要写死交给 YARN 分配。numberOfTaskSlots建议等于或略小于 CPU 核数一个 slot 一份线程资源开太多会互相抢 CPU。内存是重点# ---- 内存Flink 1.13 的 process 三级模型 ---- jobmanager.memory.process.size: 2048m taskmanager.memory.process.size: 8192m # 托管内存占比RocksDB 状态后端主要吃这块 taskmanager.memory.managed.fraction: 0.4 # 网络缓冲shuffle 密集的作业要往上抬 taskmanager.memory.network.min: 128m taskmanager.memory.network.max: 1g taskmanager.memory.network.fraction: 0.1process.size是容器总内存里面包含 Flink 自己用的堆内外内存、JVM Metaspace、JVM Overhead 三部分Flink 会自动按比例切。很多人写taskmanager.memory.process.size: 4096m然后给 YARN 申请 4G结果容器被 YARN 杀掉就是因为process.size已经是总内存YARN 那边还要额外算上 container 的 overhead。我的做法是 YARN 申请值比process.size大 10% 左右或者在 per-job 提交时直接用-Dtaskmanager.memory.process.size覆盖让它跟着 YARN 的yarn.scheduler.maximum-allocation-mb走。managed.fraction是 RocksDB 能用的托管内存比例默认 0.4状态大的作业比如按天窗口聚合调到 0.6 更稳代价是网络缓冲被挤压要看具体作业是状态重还是 shuffle 重。network.min/max是 shuffle 缓冲的上下限网络缓冲不足时报的是Insufficient number of network buffers看到这个错就把max往上抬。3.2 Checkpoint 与 Savepoint 落到 HDFS状态要能恢复checkpoint 必须落到 HDFS落本地盘在容器重启后就没了# ---- 状态后端 ---- state.backend: rocksdb state.backend.incremental: true # ---- Checkpoint ---- execution.checkpointing.interval: 60000 execution.checkpointing.mode: EXACTLY_ONCE execution.checkpointing.timeout: 10min execution.checkpointing.min-pause: 30000 execution.checkpointing.max-concurrent-checkpoints: 1 execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION # ---- 存储位置 ---- state.checkpoint-storage: filesystem state.checkpoints.dir: hdfs://cdh-cluster/flink/checkpoints state.savepoints.dir: hdfs://cdh-cluster/flink/savepoints注意execution.checkpointing.*这个前缀1.11 之前的写法是checkpointing.interval1.13 上老写法会被当成未知配置静默忽略作业照跑但永远不触发 checkpoint这种“看起来正常其实没开”的情况最难查。state.checkpoint-storage是 1.13 才有的显式配置默认走 JobManager 内存状态一大就 OOM建议显式写filesystem。state.checkpoints.dir里的cdh-cluster是你集群的 nameservice 名必须和core-site.xml里fs.defaultFS完全一致写成 IP:8020 在 HA 集群上会因为切换失效。目录权限要提前建# 提前建好目录并授权否则第一个 checkpoint 就会失败 hdfs dfs -mkdir -p /flink/checkpoints /flink/savepoints /flink/ha hdfs dfs -chown -R hadoop:hadoop /flink hdfs dfs -chmod -R 755 /flinkincremental: true只对 RocksDB 生效好处是增量上传坏处是恢复时要读更多文件状态特别小的作业可以关掉。externalized-checkpoint-retention设成RETAIN_ON_CANCELLATION是我个人的习惯取消作业时保留 checkpoint出问题还有后悔药。3.3 ZooKeeper HA 与集群 ID 的隔离生产环境 JobManager 至少两个靠 ZooKeeper 选主# ---- 高可用 ---- high-availability: zookeeper high-availability.storageDir: hdfs://cdh-cluster/flink/ha high-availability.zookeeper.quorum: node01:2181,node02:2181,node03:2181 high-availability.zookeeper.path.root: /flink high-availability.cluster-id: /flink-prod-01 high-availability.jobmanager.port: 50010high-availability.storageDir放的是 JobManager 元数据不是作业状态别和 checkpoint 目录混用。zookeeper.path.root是 znode 的根路径CDH 的 ZooKeeper 是 3.4.5没有 chroot 的额外配置路径直接写在 quorum 后面的:2181/flink也行但更推荐用path.root来隔离。cluster-id这一项最容易出事同一套 ZooKeeper 上跑多套 Flink 集群比如测试和生产共用cluster-id必须不一样否则两个 JobManager 会互相抢主表现是作业莫名其妙重启、Web UI 一会儿能开一会儿打不开。我一般按环境命名/flink-prod-01、/flink-test-01一眼能区分。注意改完flink-conf.yaml一定要在每台节点上都同步一份用 scp 或者配置管理工具都行。只在 node01 改、其他节点还是老配置是分布式的经典翻车点。4. 提交到 YARNsession、per-job、application 三种模式的参数取舍CDH 6.3.2 的集群资源都归 YARN 管Flink 的 standalone 模式只是过渡正式跑还是得提交到 YARN。flink 1.13.1 提供三种提交模式选错了不是跑不起来而是资源利用率很难看、作业之间互相干扰。4.1 Session 模式常驻集群适合反复提交的小作业Session 模式先起一个常驻的 Flink 集群后面所有作业共享这批 TaskManager# -n 起的 TaskManager 数量-s 每个 TM 的 slot 数 # -jm/-tm 覆盖配置文件里的 process.size-nm 是 YARN 上显示的应用名 /opt/flink/bin/yarn-session.sh \ -n 4 \ -s 4 \ -jm 2048m \ -tm 8192m \ -nm flink-session-130 \ -qu root.flink \ -d # 提交作业到已有的 session /opt/flink/bin/flink run \ -c com.demo.stream.DemoJob \ -p 8 \ /opt/flink/jobs/demo.jar-n 4 -s 4意思是 4 个 TaskManager、每个 4 个 slot总并发能力 16-p 8覆盖了配置里的默认并行度。-qu root.flink指定 YARN 队列CDH 里默认队列如果开了容量调度不指定会排很久。-d是后台提交不加就一直挂在终端上。Session 模式的坑在于资源隔离A 作业把 slot 占满B 作业就得排队而且一个 TaskManager 挂掉会影响上面所有作业适合开发调试和一堆小作业混跑的场景。提交完用yarn application -list能看到flink-session-130这个应用Web UI 走http://node01:8081。4.2 Per-job 与 Application 模式一个作业一套资源Per-job 模式每次提交都单独起一个集群作业结束资源自动释放/opt/flink/bin/flink run \ -t yarn-per-job \ -c com.demo.stream.DemoJob \ -Djobmanager.memory.process.size2048m \ -Dtaskmanager.memory.process.size8192m \ -Dparallelism.default8 \ -Dyarn.application.namedemo-perjob \ -Dyarn.application-attempts3 \ /opt/flink/jobs/demo.jar-t yarn-per-job是 1.13 里指定部署目标的写法-D后面跟的配置优先级高于flink-conf.yaml适合按作业调参。yarn.application-attempts: 3表示失败后最多重试 3 次配合 checkpoint 能自动恢复排查时这个值别设太大否则作业反复重启但一直失败日志会被冲掉。Application 模式把main()放到集群里执行客户端不再参与适合长时间跑的生产作业# jar 必须放在 HDFS 这类集群可访问的路径 hdfs dfs -put -f /opt/flink/jobs/demo.jar /flink/jars/demo.jar /opt/flink/bin/flink run-application \ -t yarn-application \ -c com.demo.stream.DemoJob \ -Dyarn.application.namedemo-app \ -Dparallelism.default8 \ hdfs://cdh-cluster/flink/jars/demo.jarapplication 模式的好处是客户端断开不影响作业网络抖动也不会把提交链路搞断。代价是调试麻烦一点日志要去 YARN 那边看yarn logs -applicationId application_xxx是常用命令。如果不想每次都从客户端上传 Flink 的 dist 包配一下yarn.provided.lib.dirs指向 HDFS 上的 dist 目录提交能快十几秒。4.3 Kerberos 与 Hadoop 配置怎么进容器CDH 6.3.2 默认开 Kerberos不开这个开关作业提交到 YARN 后会卡在 HDFS 认证上# ---- Kerberos ---- security.kerberos.login.use-ticket-cache: false security.kerberos.login.keytab: /etc/security/keytabs/flink.keytab security.kerberos.login.principal: flinkDEMO.COM security.kerberos.login.contexts: Client,KafkaClient security.kerberos.access.hadoopFileSystems: hdfs://cdh-cluster # ---- 让容器里也带上 Hadoop 配置 ---- containerized.master.env.HADOOP_CONF_DIR: /etc/hadoop/conf containerized.taskmanager.env.HADOOP_CONF_DIR: /etc/hadoop/conf yarn.ship-files: /etc/hadoop/confuse-ticket-cache: false配合 keytab 是容器环境的标准写法因为容器里没有kinit出来的缓存文件。security.kerberos.login.contexts里的KafkaClient只在作业要连 Kafka 时加不加会报找不到 login context。containerized.master.env.*和containerized.taskmanager.env.*这两个前缀是往 YARN 容器里注入环境变量的入口Hadoop 配置目录靠它带进去比在提交命令里拼参数清爽得多。如果 keytab 文件权限不对会出现GSSException: No valid credentials providedchmod 400加属主hadoop是最低要求。另外 Kerberos 对时间敏感节点间时间偏差超过 5 分钟就认证失败ntpdate或者 chrony 得保证同步。5. 排查清单装完起不来的五个现场与它们的解法这一节列的五个现场是我在这套组合上真实撞过的按出现频率从高到低排。每条按“现象 → 原因 → 解决”写照着对号入座就行。5.1 类路径相关的两个高频翻车现象一提交到 YARN 的作业在客户端阶段正常一进容器就抛java.lang.NoClassDefFoundError: org/apache/hadoop/fs/FileSystem堆栈里能看到FileSystem.get。原因flink 1.13.1 的发行包不含 Hadoop 依赖只有HADOOP_CONF_DIR进去了Hadoop 的 jar 没进去。很多人只在提交节点上export HADOOP_CLASSPATH但 YARN 容器是另一套环境不继承这个变量。解决确认flink-conf.yaml里containerized.master.env.HADOOP_CONF_DIR和containerized.taskmanager.env.HADOOP_CONF_DIR都写了如果还是不行退回到往$FLINK_HOME/lib塞 shaded Hadoop uber jar 的方案把HADOOP_CLASSPATH去掉避免两份类打架。改完必须重启 session 集群改配置文件对已运行的集群不生效。现象二用 Hive connector 读 CDH 6.3.2 的 Hive 2.1.1 表抛NoSuchMethodError或者NoClassDefFoundError: org/apache/calcite/...堆栈里能看到hive-exec。原因Flink 1.13.1 的 Hive connector 官方覆盖 1.2.1、2.3.6、3.1.2 三个版本CDH 的 2.1.1 不在列表里。直接用hive-exec-2.1.1.jar会和 Flink 内置的 Calcite 版本冲突两边的 Calcite 不是同一版。解决优先用带 shaded 的flink-sql-connector-hive-2.3.6_2.11-1.13.1.jar别再往lib里手塞hive-exec如果一定要用 2.1.1就把hive-exec里org.apache.calcite包排除掉再打包。这套组合能跑但 Hive 2.1.1 特有的 UDF 可能不兼容用到就得单独适配提前评估工作量。5.2 网络与注册UI 上看不到 TaskManager现象JobManager 的 Web UI 打得开但 TaskManager 数量是 0flink list能连上集群但提交的作业一直INITIALIZING。原因TaskManager 注册时用了127.0.0.1。standalone 多机部署时taskmanager.host没写Flink 会自己猜一个网卡地址猜错就注册回环。另一种可能是jobmanager.rpc.address用了主机名但/etc/hosts各节点不一致解析到不同机器上。解决standalone 模式每台 TaskManager 节点的flink-conf.yaml里显式写taskmanager.host: 本机主机名或IP同时检查jobmanager.rpc.address在所有节点上解析结果是否一致。YARN 模式下反而不要写taskmanager.host交给 YARN 分配写了会覆盖成固定值多容器场景直接冲突。5.3 资源与权限容器被杀和 checkpoint 写不进去现象作业跑几分钟后 TaskManager 消失YARN 日志里是Container killed by YARN for exceeding memory limits. 4.9 GB of 4 GB physical memory used。原因taskmanager.memory.process.size设成 4096mYARN 申请值也按 4G 报但process.size之外还有 JVM 自身的开销线程栈、Direct Memory、Metaspace 余量实际占用一定超过 4GYARN 的监控一超就杀。解决YARN 申请值比process.size高 10%15%或者直接用-Dtaskmanager.memory.process.size让它小于 YARN 容器的可用内存。调managed.fraction也能缓解但那是把状态内存让出来治标不治本。另外yarn.nodemanager.resource.memory-mb要够不然申请本身就会失败。现象checkpoint 全部失败日志里是FileSystemException或者Permission denied: userhadoop, accessWRITE, inode/flink/checkpoints。原因两种可能一是state.checkpoints.dir写成了hdfs://node01:8020/...而集群是 HA 的端口和 nameservice 对不上二是目录属主不是运行用户/flink建在了 root 名下。解决路径统一用 nameservice 名和fs.defaultFS保持一致目录用hdfs dfs -chown -R hadoop:hadoop /flink授权并在flink-conf.yaml里确认没有残留 IP 写法。改完先手动跑一次hdfs dfs -touchz /flink/checkpoints/.test验证权限比等作业报错快得多。6. 装完怎么验证WordCount 跑通到 Savepoint 恢复的全链路自检装完不验证等于没装。我一般不看“进程起来了没”而是走一遍提交、跑数、checkpoint、savepoint、恢复的完整链路五步全过才算这套环境可用。第一步先跑一个带状态的 demo 作业不要用无状态的 WordCount测不出 checkpoint 的问题。提交后立刻看 UI 的 Checkpoints 页签Completed数量应该在两三个周期内开始增长# 提交一个带状态的自检作业观察 checkpoint 是否正常触发 /opt/flink/bin/flink run \ -t yarn-per-job \ -c com.demo.stream.SelfCheckJob \ -Dexecution.checkpointing.interval10000 \ -Dyarn.application.nameflink-selfcheck \ /opt/flink/jobs/selfcheck.jar # 拿到 jobId 后确认状态是 RUNNING /opt/flink/bin/flink list -t yarn-per-job -Dyarn.application.idapplication_xxx第二步去 HDFS 上确认 checkpoint 真的落盘文件的修改时间要在一分钟以内hdfs dfs -ls hdfs://cdh-cluster/flink/checkpoints/jobId/ | tail -5第三步手动打一个 savepoint这一步会真正走一遍状态快照的完整路径比 checkpoint 严格/opt/flink/bin/flink savepoint jobId hdfs://cdh-cluster/flink/savepoints -t yarn-per-job -Dyarn.application.idapplication_xxx第四步取消作业并从 savepoint 恢复这一步是整套验证里最有价值的能一次性暴露状态后端、HDFS 权限、序列化三类问题# 取消并触发 savepoint /opt/flink/bin/flink cancel -s hdfs://cdh-cluster/flink/savepoints jobId # 从 savepoint 恢复 /opt/flink/bin/flink run -t yarn-per-job -s hdfs://cdh-cluster/flink/savepoints/savepoint-xxxx \ -c com.demo.stream.SelfCheckJob /opt/flink/jobs/selfcheck.jar第五步对一张表做验收检查项期望结果不通过时的方向YARN 应用状态提交后 30 秒内 RUNNING看 ResourceManager 日志和队列资源TaskManager 数量与-n或并行度匹配查taskmanager.host和端口Checkpoint数量持续增长大小稳定查 HDFS 权限和 nameserviceSavepoint生成成功且可恢复查状态后端和序列化器Web UI 指标有背压和吞吐数据查metrics.reporter配置恢复成功之后作业的numRestarts会归零重新计数状态里累计的数值应该接着之前的走而不是从 0 开始——这一点最能证明 savepoint 是真的生效了。从那以后我每次搭这套环境都强制走一遍 savepoint 恢复再交付哪怕只是测试集群。因为 checkpoint 能过、savepoint 恢复不了的情况太常见了往往卡在状态序列化或者 HDFS 路径上等到生产出事再发现代价就不是半小时能补回来的。这套 CDH 6.3.2 加 flink 1.13.1 的组合本身没什么坑坑都在版本对齐和配置项的前后缀上照着上面这一遍走下来基本能落地。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/24 4:34:19 · 阅读 1.2k CDH 6.3.2 集成 Flink 1.13.1：版本对齐与 YARN 排障
简介面向在 CDH 6.3.2 集群上部署 Flink 1.13.1 的大数据运维与开发人员这份资源提供一套可用于 Parcel 分发和 YARN 提交的部署文件集合重点解决离线环境组件不全、版本匹配与依赖缺失等问题。压缩包共 5 个文件约 299.52MB主要由两个 jar、一个 parcel、一个 sha 校验文件和一个 json 元数据组成jar 覆盖 YARN 客户端与核心运行库parcel 承载 Scala 2.11、el7 环境的二进制发行版sha 与 json 则分别用于完整性校验和分发元信息。目前已有 3498 人学习。对于需要将 Flink 接入现有 Hadoop 生态的团队这套文件能帮助快速完成组件准备、版本核对与集群侧分发减少因包不完整或校验失败导致的反复排查其文件组合也便于在离线集群中核对依赖关系、统一部署口径并为后续作业提交提供一致的运行环境适合作为该平台部署的基础物料。1. 版本先对齐CDH 6.3.2 上跑 flink 1.13.1 的前置判断上周给一套 CDH 6.3.2 的集群加流计算层运维图省事从归档站直接拖了flink-1.13.1-bin-scala_2.12.tgz下来解压作业一提交就死在NoSuchMethodError: scala.Predef$.refArrayOps日志翻半天没有任何有用信息。这不是 Flink 坏了是 Scala 版本对不上——CDH 6.3.2 整条生态线是 Scala 2.11flink 1.13.1 的二进制包分 2.11 和 2.12 两套装错了连 Hello World 都跑不起来。更隐蔽的是 Hadoop从 1.11 起 Flink 的发行包不再内置 Hadoop 依赖CDH 6.3.2 又自带 Hadoop 3.0.0套用网上那些“解压即用”的老教程大概率卡在NoClassDefFoundError: org/apache/hadoop/fs/FileSystem。这篇按我实际落地的顺序走一遍选包、配环境、写flink-conf.yaml、提交 YARN、出事怎么救。2. 装前备料JDK、Scala 2.11 与 CDH 客户端的路径对齐装 Flink 最容易被跳过的一步就是“对版本”很多人上来就解压出问题再回头查来回折腾两小时。这一节把三件事定死JDK 用哪个、Scala 包怎么选、Hadoop 客户端从哪里来。这三样定完后面的配置基本是填空。2.1 一张表定死四个组件的版本先把 CDH 6.3.2 自带的版本和 flink 1.13.1 的要求摆在一起看冲突点一眼就能看出来组件CDH 6.3.2 自带flink 1.13.1 的要求落地结论JDK1.8.0 系列Java 8 主线支持Java 11 属实验态直接用集群自带 8别单独升级Scala2.11.x发行包分 scala_2.11 / scala_2.12必须选-bin-scala_2.11那个包Hadoop3.0.0二进制包不再内置 Hadoop 依赖靠HADOOP_CLASSPATH兜Hive2.1.1connector 官方覆盖 1.2.1 / 2.3.6 / 3.1.22.1.1 属于要绕的版本ZooKeeper3.4.53.4 及以上可用直接复用集群的不再单独装状态后端—RocksDB 依赖 glibc 2.17CentOS 7 基线满足不用额外装表里最后一行值得多说一句RocksDB 是 native 库libflink-runtime里那份.so对 glibc 版本有要求CentOS 6 的老机器上会抛UnsatisfiedLinkError这种机器别硬上换节点。至于 HiveCDH 6.3.2 的 2.1.1 不在官方列表里后面第 5 章会单独讲怎么处理。2.2 上传、解压与运行用户不要用 root 跑 Flink 进程CDH 集群里一般已经有hadoop这个服务账号直接复用最省事HDFS 权限也好理。包传到一台能连 HDFS 和 YARN 的边缘节点上解压到一个固定路径# 上传后解压到 /opt目录名统一改成 flink方便后面写环境变量 tar -zxvf flink-1.13.1-bin-scala_2.11.tgz -C /opt mv /opt/flink-1.13.1 /opt/flink # 属主给集群统一的服务账号避免 HDFS 写 checkpoint 时 Permission denied chown -R hadoop:hadoop /opt/flink chmod 755 /opt/flink # 确认包里的 Scala 版本没拿错 ls /opt/flink/lib | grep flink-distflink-dist_2.11-1.13.1.jar这个文件名里的2.11就是最后一道校验如果这里出现的是2.12不用往下走了重新下包。解压路径建议所有节点保持一致/opt/flink这种固定路径能省掉后面一堆路径写错的排查。2.3 环境变量与 HADOOP_CLASSPATH 的两种给法flink 1.13.1 的发行包不带 Hadoop所以要主动把 Hadoop 的类喂给它。业内有两种做法我一般优先选第一种因为不引入第二份 Hadoop 类冲突面小。第一种是让 Flink 用集群自己的 Hadoop 客户端cat /etc/profile.d/flink.sh EOF # Flink 主目录 export FLINK_HOME/opt/flink # CDH parcel 的解压目录因安装方式而异先探测再赋值避免写死 export CDH_HOME$(ls -d /opt/*/parcels/CDH 2/dev/null | head -1) # Hadoop 客户端与配置目录 export HADOOP_HOME$CDH_HOME/lib/hadoop export HADOOP_CONF_DIR/etc/hadoop/conf export YARN_CONF_DIR/etc/hadoop/conf export HBASE_CONF_DIR/etc/hbase/conf # 关键把 hadoop classpath 交给 Flink缺这行就是 NoClassDefFoundError export HADOOP_CLASSPATH$(hadoop classpath) export PATH$PATH:$FLINK_HOME/bin EOF source /etc/profile.d/flink.sh # 自检这两条都得有输出 echo $HADOOP_CONF_DIR hadoop classpath | head -c 200HADOOP_CLASSPATH的本质是把hadoop classpath的输出一长串 jar 路径用冒号分隔挂到 Flink 进程的 classpath 上。HADOOP_CONF_DIR则决定 Flink 能不能读到core-site.xml、hdfs-site.xml、yarn-site.xml提交到 YARN 时这些配置文件会被打进容器缺了就是连不上 ResourceManager。hadoop classpath命令本身依赖 CDH 的 alternatives 链接如果提示 command not found先用绝对路径$CDH_HOME/bin/hadoop classpath试一次能用就说明是 PATH 问题。第二种做法是往lib里塞一个 shaded 的 Hadoop uber jar好处是HADOOP_CLASSPATH不用管坏处是这份 jar 的 Hadoop 版本和 CDH 的 3.0.0 不完全一致某些 HDFS 新特性会踩空。两种方式不要同时用双份 Hadoop 类在 classpath 上打架报错信息会非常迷惑——这是血泪经验我见过一次ClassCastException: org.apache.hadoop.hdfs.DistributedFileSystem cannot be cast to ...查了一整天才定位到是两边都有。注意HADOOP_CLASSPATH内容很长某些场景会顶到命令行长度上限如果启动时报Argument list too long改用env.java.classpath或者只保留HADOOP_CONF_DIR把 Hadoop jar 交给 YARN 容器自带的 classpath。3. 配置 flink-conf.yaml内存、状态后端与 HA 三段写全conf/flink-conf.yaml是这套环境的黑匣子跑不起来十有八九能从这儿找到线索。flink 1.13.1 用的是 process 三级内存模型和 1.9 之前那套taskmanager.heap.size完全不是一回事照抄老配置会直接被忽略甚至报错。下面按基础、内存、状态与 HA 三段拆开写。3.1 基础项与内存模型的四个数先把通信和并行度定下来# ---- 基础通信 ---- jobmanager.rpc.address: node01 jobmanager.rpc.port: 6123 # 绑定 0.0.0.0 是为了让外部能访问 8081 的 Web UI jobmanager.bind-host: 0.0.0.0 taskmanager.bind-host: 0.0.0.0 # ---- 并行度 ---- parallelism.default: 4 taskmanager.numberOfTaskSlots: 4jobmanager.rpc.address写主机名还是 IP取决于集群/etc/hosts是否统一我一般写主机名方便后面换节点。taskmanager.host这一项在 standalone 多机部署时必须逐台改成自己否则 TaskManager 会拿127.0.0.1去注册JobManager 永远看不到 slot但在 YARN 模式下不要写死交给 YARN 分配。numberOfTaskSlots建议等于或略小于 CPU 核数一个 slot 一份线程资源开太多会互相抢 CPU。内存是重点# ---- 内存Flink 1.13 的 process 三级模型 ---- jobmanager.memory.process.size: 2048m taskmanager.memory.process.size: 8192m # 托管内存占比RocksDB 状态后端主要吃这块 taskmanager.memory.managed.fraction: 0.4 # 网络缓冲shuffle 密集的作业要往上抬 taskmanager.memory.network.min: 128m taskmanager.memory.network.max: 1g taskmanager.memory.network.fraction: 0.1process.size是容器总内存里面包含 Flink 自己用的堆内外内存、JVM Metaspace、JVM Overhead 三部分Flink 会自动按比例切。很多人写taskmanager.memory.process.size: 4096m然后给 YARN 申请 4G结果容器被 YARN 杀掉就是因为process.size已经是总内存YARN 那边还要额外算上 container 的 overhead。我的做法是 YARN 申请值比process.size大 10% 左右或者在 per-job 提交时直接用-Dtaskmanager.memory.process.size覆盖让它跟着 YARN 的yarn.scheduler.maximum-allocation-mb走。managed.fraction是 RocksDB 能用的托管内存比例默认 0.4状态大的作业比如按天窗口聚合调到 0.6 更稳代价是网络缓冲被挤压要看具体作业是状态重还是 shuffle 重。network.min/max是 shuffle 缓冲的上下限网络缓冲不足时报的是Insufficient number of network buffers看到这个错就把max往上抬。3.2 Checkpoint 与 Savepoint 落到 HDFS状态要能恢复checkpoint 必须落到 HDFS落本地盘在容器重启后就没了# ---- 状态后端 ---- state.backend: rocksdb state.backend.incremental: true # ---- Checkpoint ---- execution.checkpointing.interval: 60000 execution.checkpointing.mode: EXACTLY_ONCE execution.checkpointing.timeout: 10min execution.checkpointing.min-pause: 30000 execution.checkpointing.max-concurrent-checkpoints: 1 execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION # ---- 存储位置 ---- state.checkpoint-storage: filesystem state.checkpoints.dir: hdfs://cdh-cluster/flink/checkpoints state.savepoints.dir: hdfs://cdh-cluster/flink/savepoints注意execution.checkpointing.*这个前缀1.11 之前的写法是checkpointing.interval1.13 上老写法会被当成未知配置静默忽略作业照跑但永远不触发 checkpoint这种“看起来正常其实没开”的情况最难查。state.checkpoint-storage是 1.13 才有的显式配置默认走 JobManager 内存状态一大就 OOM建议显式写filesystem。state.checkpoints.dir里的cdh-cluster是你集群的 nameservice 名必须和core-site.xml里fs.defaultFS完全一致写成 IP:8020 在 HA 集群上会因为切换失效。目录权限要提前建# 提前建好目录并授权否则第一个 checkpoint 就会失败 hdfs dfs -mkdir -p /flink/checkpoints /flink/savepoints /flink/ha hdfs dfs -chown -R hadoop:hadoop /flink hdfs dfs -chmod -R 755 /flinkincremental: true只对 RocksDB 生效好处是增量上传坏处是恢复时要读更多文件状态特别小的作业可以关掉。externalized-checkpoint-retention设成RETAIN_ON_CANCELLATION是我个人的习惯取消作业时保留 checkpoint出问题还有后悔药。3.3 ZooKeeper HA 与集群 ID 的隔离生产环境 JobManager 至少两个靠 ZooKeeper 选主# ---- 高可用 ---- high-availability: zookeeper high-availability.storageDir: hdfs://cdh-cluster/flink/ha high-availability.zookeeper.quorum: node01:2181,node02:2181,node03:2181 high-availability.zookeeper.path.root: /flink high-availability.cluster-id: /flink-prod-01 high-availability.jobmanager.port: 50010high-availability.storageDir放的是 JobManager 元数据不是作业状态别和 checkpoint 目录混用。zookeeper.path.root是 znode 的根路径CDH 的 ZooKeeper 是 3.4.5没有 chroot 的额外配置路径直接写在 quorum 后面的:2181/flink也行但更推荐用path.root来隔离。cluster-id这一项最容易出事同一套 ZooKeeper 上跑多套 Flink 集群比如测试和生产共用cluster-id必须不一样否则两个 JobManager 会互相抢主表现是作业莫名其妙重启、Web UI 一会儿能开一会儿打不开。我一般按环境命名/flink-prod-01、/flink-test-01一眼能区分。注意改完flink-conf.yaml一定要在每台节点上都同步一份用 scp 或者配置管理工具都行。只在 node01 改、其他节点还是老配置是分布式的经典翻车点。4. 提交到 YARNsession、per-job、application 三种模式的参数取舍CDH 6.3.2 的集群资源都归 YARN 管Flink 的 standalone 模式只是过渡正式跑还是得提交到 YARN。flink 1.13.1 提供三种提交模式选错了不是跑不起来而是资源利用率很难看、作业之间互相干扰。4.1 Session 模式常驻集群适合反复提交的小作业Session 模式先起一个常驻的 Flink 集群后面所有作业共享这批 TaskManager# -n 起的 TaskManager 数量-s 每个 TM 的 slot 数 # -jm/-tm 覆盖配置文件里的 process.size-nm 是 YARN 上显示的应用名 /opt/flink/bin/yarn-session.sh \ -n 4 \ -s 4 \ -jm 2048m \ -tm 8192m \ -nm flink-session-130 \ -qu root.flink \ -d # 提交作业到已有的 session /opt/flink/bin/flink run \ -c com.demo.stream.DemoJob \ -p 8 \ /opt/flink/jobs/demo.jar-n 4 -s 4意思是 4 个 TaskManager、每个 4 个 slot总并发能力 16-p 8覆盖了配置里的默认并行度。-qu root.flink指定 YARN 队列CDH 里默认队列如果开了容量调度不指定会排很久。-d是后台提交不加就一直挂在终端上。Session 模式的坑在于资源隔离A 作业把 slot 占满B 作业就得排队而且一个 TaskManager 挂掉会影响上面所有作业适合开发调试和一堆小作业混跑的场景。提交完用yarn application -list能看到flink-session-130这个应用Web UI 走http://node01:8081。4.2 Per-job 与 Application 模式一个作业一套资源Per-job 模式每次提交都单独起一个集群作业结束资源自动释放/opt/flink/bin/flink run \ -t yarn-per-job \ -c com.demo.stream.DemoJob \ -Djobmanager.memory.process.size2048m \ -Dtaskmanager.memory.process.size8192m \ -Dparallelism.default8 \ -Dyarn.application.namedemo-perjob \ -Dyarn.application-attempts3 \ /opt/flink/jobs/demo.jar-t yarn-per-job是 1.13 里指定部署目标的写法-D后面跟的配置优先级高于flink-conf.yaml适合按作业调参。yarn.application-attempts: 3表示失败后最多重试 3 次配合 checkpoint 能自动恢复排查时这个值别设太大否则作业反复重启但一直失败日志会被冲掉。Application 模式把main()放到集群里执行客户端不再参与适合长时间跑的生产作业# jar 必须放在 HDFS 这类集群可访问的路径 hdfs dfs -put -f /opt/flink/jobs/demo.jar /flink/jars/demo.jar /opt/flink/bin/flink run-application \ -t yarn-application \ -c com.demo.stream.DemoJob \ -Dyarn.application.namedemo-app \ -Dparallelism.default8 \ hdfs://cdh-cluster/flink/jars/demo.jarapplication 模式的好处是客户端断开不影响作业网络抖动也不会把提交链路搞断。代价是调试麻烦一点日志要去 YARN 那边看yarn logs -applicationId application_xxx是常用命令。如果不想每次都从客户端上传 Flink 的 dist 包配一下yarn.provided.lib.dirs指向 HDFS 上的 dist 目录提交能快十几秒。4.3 Kerberos 与 Hadoop 配置怎么进容器CDH 6.3.2 默认开 Kerberos不开这个开关作业提交到 YARN 后会卡在 HDFS 认证上# ---- Kerberos ---- security.kerberos.login.use-ticket-cache: false security.kerberos.login.keytab: /etc/security/keytabs/flink.keytab security.kerberos.login.principal: flinkDEMO.COM security.kerberos.login.contexts: Client,KafkaClient security.kerberos.access.hadoopFileSystems: hdfs://cdh-cluster # ---- 让容器里也带上 Hadoop 配置 ---- containerized.master.env.HADOOP_CONF_DIR: /etc/hadoop/conf containerized.taskmanager.env.HADOOP_CONF_DIR: /etc/hadoop/conf yarn.ship-files: /etc/hadoop/confuse-ticket-cache: false配合 keytab 是容器环境的标准写法因为容器里没有kinit出来的缓存文件。security.kerberos.login.contexts里的KafkaClient只在作业要连 Kafka 时加不加会报找不到 login context。containerized.master.env.*和containerized.taskmanager.env.*这两个前缀是往 YARN 容器里注入环境变量的入口Hadoop 配置目录靠它带进去比在提交命令里拼参数清爽得多。如果 keytab 文件权限不对会出现GSSException: No valid credentials providedchmod 400加属主hadoop是最低要求。另外 Kerberos 对时间敏感节点间时间偏差超过 5 分钟就认证失败ntpdate或者 chrony 得保证同步。5. 排查清单装完起不来的五个现场与它们的解法这一节列的五个现场是我在这套组合上真实撞过的按出现频率从高到低排。每条按“现象 → 原因 → 解决”写照着对号入座就行。5.1 类路径相关的两个高频翻车现象一提交到 YARN 的作业在客户端阶段正常一进容器就抛java.lang.NoClassDefFoundError: org/apache/hadoop/fs/FileSystem堆栈里能看到FileSystem.get。原因flink 1.13.1 的发行包不含 Hadoop 依赖只有HADOOP_CONF_DIR进去了Hadoop 的 jar 没进去。很多人只在提交节点上export HADOOP_CLASSPATH但 YARN 容器是另一套环境不继承这个变量。解决确认flink-conf.yaml里containerized.master.env.HADOOP_CONF_DIR和containerized.taskmanager.env.HADOOP_CONF_DIR都写了如果还是不行退回到往$FLINK_HOME/lib塞 shaded Hadoop uber jar 的方案把HADOOP_CLASSPATH去掉避免两份类打架。改完必须重启 session 集群改配置文件对已运行的集群不生效。现象二用 Hive connector 读 CDH 6.3.2 的 Hive 2.1.1 表抛NoSuchMethodError或者NoClassDefFoundError: org/apache/calcite/...堆栈里能看到hive-exec。原因Flink 1.13.1 的 Hive connector 官方覆盖 1.2.1、2.3.6、3.1.2 三个版本CDH 的 2.1.1 不在列表里。直接用hive-exec-2.1.1.jar会和 Flink 内置的 Calcite 版本冲突两边的 Calcite 不是同一版。解决优先用带 shaded 的flink-sql-connector-hive-2.3.6_2.11-1.13.1.jar别再往lib里手塞hive-exec如果一定要用 2.1.1就把hive-exec里org.apache.calcite包排除掉再打包。这套组合能跑但 Hive 2.1.1 特有的 UDF 可能不兼容用到就得单独适配提前评估工作量。5.2 网络与注册UI 上看不到 TaskManager现象JobManager 的 Web UI 打得开但 TaskManager 数量是 0flink list能连上集群但提交的作业一直INITIALIZING。原因TaskManager 注册时用了127.0.0.1。standalone 多机部署时taskmanager.host没写Flink 会自己猜一个网卡地址猜错就注册回环。另一种可能是jobmanager.rpc.address用了主机名但/etc/hosts各节点不一致解析到不同机器上。解决standalone 模式每台 TaskManager 节点的flink-conf.yaml里显式写taskmanager.host: 本机主机名或IP同时检查jobmanager.rpc.address在所有节点上解析结果是否一致。YARN 模式下反而不要写taskmanager.host交给 YARN 分配写了会覆盖成固定值多容器场景直接冲突。5.3 资源与权限容器被杀和 checkpoint 写不进去现象作业跑几分钟后 TaskManager 消失YARN 日志里是Container killed by YARN for exceeding memory limits. 4.9 GB of 4 GB physical memory used。原因taskmanager.memory.process.size设成 4096mYARN 申请值也按 4G 报但process.size之外还有 JVM 自身的开销线程栈、Direct Memory、Metaspace 余量实际占用一定超过 4GYARN 的监控一超就杀。解决YARN 申请值比process.size高 10%15%或者直接用-Dtaskmanager.memory.process.size让它小于 YARN 容器的可用内存。调managed.fraction也能缓解但那是把状态内存让出来治标不治本。另外yarn.nodemanager.resource.memory-mb要够不然申请本身就会失败。现象checkpoint 全部失败日志里是FileSystemException或者Permission denied: userhadoop, accessWRITE, inode/flink/checkpoints。原因两种可能一是state.checkpoints.dir写成了hdfs://node01:8020/...而集群是 HA 的端口和 nameservice 对不上二是目录属主不是运行用户/flink建在了 root 名下。解决路径统一用 nameservice 名和fs.defaultFS保持一致目录用hdfs dfs -chown -R hadoop:hadoop /flink授权并在flink-conf.yaml里确认没有残留 IP 写法。改完先手动跑一次hdfs dfs -touchz /flink/checkpoints/.test验证权限比等作业报错快得多。6. 装完怎么验证WordCount 跑通到 Savepoint 恢复的全链路自检装完不验证等于没装。我一般不看“进程起来了没”而是走一遍提交、跑数、checkpoint、savepoint、恢复的完整链路五步全过才算这套环境可用。第一步先跑一个带状态的 demo 作业不要用无状态的 WordCount测不出 checkpoint 的问题。提交后立刻看 UI 的 Checkpoints 页签Completed数量应该在两三个周期内开始增长# 提交一个带状态的自检作业观察 checkpoint 是否正常触发 /opt/flink/bin/flink run \ -t yarn-per-job \ -c com.demo.stream.SelfCheckJob \ -Dexecution.checkpointing.interval10000 \ -Dyarn.application.nameflink-selfcheck \ /opt/flink/jobs/selfcheck.jar # 拿到 jobId 后确认状态是 RUNNING /opt/flink/bin/flink list -t yarn-per-job -Dyarn.application.idapplication_xxx第二步去 HDFS 上确认 checkpoint 真的落盘文件的修改时间要在一分钟以内hdfs dfs -ls hdfs://cdh-cluster/flink/checkpoints/jobId/ | tail -5第三步手动打一个 savepoint这一步会真正走一遍状态快照的完整路径比 checkpoint 严格/opt/flink/bin/flink savepoint jobId hdfs://cdh-cluster/flink/savepoints -t yarn-per-job -Dyarn.application.idapplication_xxx第四步取消作业并从 savepoint 恢复这一步是整套验证里最有价值的能一次性暴露状态后端、HDFS 权限、序列化三类问题# 取消并触发 savepoint /opt/flink/bin/flink cancel -s hdfs://cdh-cluster/flink/savepoints jobId # 从 savepoint 恢复 /opt/flink/bin/flink run -t yarn-per-job -s hdfs://cdh-cluster/flink/savepoints/savepoint-xxxx \ -c com.demo.stream.SelfCheckJob /opt/flink/jobs/selfcheck.jar第五步对一张表做验收检查项期望结果不通过时的方向YARN 应用状态提交后 30 秒内 RUNNING看 ResourceManager 日志和队列资源TaskManager 数量与-n或并行度匹配查taskmanager.host和端口Checkpoint数量持续增长大小稳定查 HDFS 权限和 nameserviceSavepoint生成成功且可恢复查状态后端和序列化器Web UI 指标有背压和吞吐数据查metrics.reporter配置恢复成功之后作业的numRestarts会归零重新计数状态里累计的数值应该接着之前的走而不是从 0 开始——这一点最能证明 savepoint 是真的生效了。从那以后我每次搭这套环境都强制走一遍 savepoint 恢复再交付哪怕只是测试集群。因为 checkpoint 能过、savepoint 恢复不了的情况太常见了往往卡在状态序列化或者 HDFS 路径上等到生产出事再发现代价就不是半小时能补回来的。这套 CDH 6.3.2 加 flink 1.13.1 的组合本身没什么坑坑都在版本对齐和配置项的前后缀上照着上面这一遍走下来基本能落地。希望帮到你。本文还有配套的精品资源点击获取