CDH 6.3.2 集成 Flink 1.16.2：Parcel 与 CSD 部署实战

简介面向需要在 CDH6.3.2 集群中集成 Flink 的大数据运维与开发人员这份资源提供 Flink-1.16.2 的完整离线安装包省去自行编译打包的繁琐环节可直接用于 Parcel 分发与 CSD 服务描述配置适合具备一定 CDH 集群管理经验的中高级用户。压缩包共 7 个文件约 518.32MB包含 Parcel 主包及其校验文件、CSD 描述文件、Flink 与 Flink on YARN 相关 jar以及 shaded Hadoop 依赖和 manifest 清单覆盖从分发、校验到服务集成的关键环节。已有 974 人学习下载说明该版本组合在实际生产环境中具有一定参考价值。借助这套包读者可快速完成 Flink 1.16.2 与 CDH6.3.2 的版本适配减少依赖冲突排查时间并结合配套部署教程理清 Parcel 放置、CSD 上传与 YARN 提交的完整链路为后续流批一体任务开发打下环境基础。1. 为什么有人非要把 Flink 1.16.2 塞进 CDH 6.3.2上周帮一个做实时数仓的朋友收拾烂摊子集群是 CDH 6.3.2Hadoop 3.0.0、Hive 2.1.1、ZooKeeper 3.4.5 一整套跑得好好的业务方突然要上 Flink 做实时链路运维第一反应是「再开一套 standalone 集群」。结果两套环境元数据对不上、Kerberos 两套 principal、监控两套面板半夜告警互相甩锅。最后绕回原点能不能把 Flink 1.16.2 做成 parcel 直接托管进 CDH 6.3.2让 Cloudera Manager 统一管起来。这个资源包解决的就是这件事。它不是 Flink 官方发行版而是一份面向 CDH 6.3.2 的 Flink 1.16.2 集成产物核心是 parcels 目录加 csd 描述文件让 Flink 以「受管服务」的身份出现在 CM 里。适合两类人一是被要求「不新增独立集群」的运维二是需要在同一套 Hadoop 生态里跑 Flink SQL 和 DataStream 作业的开发。下面按我实际拆包、部署、踩坑的顺序讲参数和路径都落到能照抄的程度。2. 拆开 parcels 与 csd这份集成包到底装了什么2.1 parcel 与 csd 的分工先搞清楚再动手CDH 的扩展机制分两层很多人第一次做集成会把两者混为一谈。parcel 是「二进制分发单元」本质是一个带元数据的压缩包里面放 Flink 的 lib、bin、conf 模板由 CM 负责分发到各节点并解压到/opt/cloudera/parcels/下。csdCustom Service Descriptor是「服务描述单元」是一个 jar里面是 service.sdl 和若干脚本告诉 CM「这个服务有哪些角色、怎么启停、配置项长什么样、健康检查怎么跑」。两者缺一不可只有 parcelCM 认不出这是个服务只有 csd没有实际二进制可分发。集成包的结构通常是这样的FLINK-1.16.2-BIN/ ├── meta/ │ ├── parcel.json │ └── flink_env.sh └── (Flink 发行版内容bin/ lib/ conf/ ...) FLINK-1.16.2.jar # csd内含 descriptor/service.sdlparcel.json里最关键的是depends字段要声明对 CDH 版本的依赖否则 CM 在「Parcel 分配」页面会直接标红不可用。常见写法是依赖CDH且版本区间覆盖 6.3.2。这里有个血泪经验depends写太死比如精确等于 6.3.2会导致小版本升级后 parcel 失效写太松又可能在不兼容的 Hadoop 上强行激活我一般按「 6.3.0, 6.4.0」这种区间来卡。2.2 校验包完整性先看哈希和目录再谈部署拿到包别急着往 CM 里塞先在跳板机上做三件事。第一核对 sha1parcel 分发失败十有八九是传输截断第二确认 parcel 顶层目录名与parcel.json里的name一致CM 是按目录名匹配的第三检查 csd jar 里有没有descriptor/service.sdl缺了它 CM 添加服务时会报「无法识别的服务描述」。# 1. 校验哈希以实际提供的 sha 文件为准 sha1sum FLINK-1.16.2-BIN-el7.parcel cat FLINK-1.16.2-BIN-el7.parcel.sha # 2. 看 parcel 顶层结构确认目录名与 parcel.json 的 name 字段一致 tar -tf FLINK-1.16.2-BIN-el7.parcel | head -20 # 3. 检查 csd jar 内是否含服务描述 unzip -l FLINK-1.16.2.jar | grep -E service.sdl|descriptor第一段命令的输出要和 sha 文件逐字符一致注意有些包给的是 sha1 有些是 sha256别拿错算法。第二段看顶层目录正常应该只有一个FLINK-1.16.2-BIN/如果出现./前缀或多层嵌套说明打包时目录层级错了CM 解压后会找不到meta/parcel.json。第三段是很多人忽略的csd jar 里如果只有 class 没有 sdl说明这是个半成品得让提供方补。2.3 放进本地 parcel 仓库并让 CM 识别CM 的 parcel 仓库可以是本地目录也可以是 http 服务生产上我倾向本地目录加 NFS 或直接放 CM 主机的/opt/cloudera/parcel-repo/。把 parcel、sha、csd jar 三个文件放进去权限给cloudera-scm:cloudera-scm然后到 CM 的「主机 → Parcel」页面点「检查新 Parcel」。# 放置文件路径以实际 parcel-repo 为准 sudo cp FLINK-1.16.2-BIN-el7.parcel \ FLINK-1.16.2-BIN-el7.parcel.sha \ FLINK-1.16.2.jar \ /opt/cloudera/parcel-repo/ sudo chown cloudera-scm:cloudera-scm /opt/cloudera/parcel-repo/FLINK-1.16.2*放完后如果 CM 页面没刷新出来先看/var/log/cloudera-scm-server/cloudera-scm-server.log里有没有 parcel 扫描报错常见是 sha 文件名不匹配CM 要求 sha 文件名是parcel名.sha多一个后缀就认不出。识别成功后先「分配」再「激活」激活前确认目标主机的磁盘余量Flink 发行版解压后大概几百 MB别卡在最后一步。3. 在 CM 里添加 Flink 服务角色分配与关键配置3.1 添加服务与角色规划parcel 激活后CM 首页「添加服务」里应该能看到 Flink 条目。点进去是角色分配页Flink 在 CDH 集成里通常拆成三类角色JobManager对应 Flink 的 master、TaskManagerworker、以及可选的 HistoryServer。小集群可以把 JobManager 和 HistoryServer 放同一台TaskManager 按数据节点铺开。角色规划有两个硬约束要提前想清楚。其一JobManager 建议单独一台或与 CM 管理节点错开它内存吃得不小默认模板给 1G 往往不够跑大状态作业容易 OOM。其二TaskManager 的 slot 数和每节点内存要匹配taskmanager.numberOfTaskSlots乘以单 slot 内存不能超过容器或物理机给的上限否则作业提交时资源申请直接失败。我一般按「每节点 4 slot、单 slot 2G、堆外留 1G」起步再按作业并行度调。3.2 必须改的配置项清单CM 里的 Flink 配置分「服务级」和「角色级」下面这张表是我每次部署都会过一遍的参数名以 Flink 1.16 的flink-conf.yaml为准CM 里对应的是同名配置项。配置项建议值说明jobmanager.rpc.addressJobManager 主机名CM 一般会自动填多 JobManager 高可用时改走 ZooKeeperjobmanager.memory.process.size2048m 起进程总内存含堆、堆外、元空间别只填 heaptaskmanager.memory.process.size4096m 起同上按 slot 数反推taskmanager.numberOfTaskSlots2~4与 CPU 核数匹配超配会互相抢parallelism.default按集群规模默认并行度作业未指定时生效state.backendrocksdb大状态必选默认 hashmap 撑不住state.checkpoints.dirHDFS 路径必须写 HDFS本地盘重启即丢high-availability.typezookeeper生产必开配合high-availability.storageDirstate.checkpoints.dir和high-availability.storageDir这两个路径要提前在 HDFS 上建好并给 Flink 运行用户写权限否则作业第一次 checkpoint 就失败日志里是Permission denied或FileNotFoundException排查起来要翻好几层。另外state.backend选 rocksdb 时要确认 parcel 里带了flink-statebackend-rocksdb的 jar有些精简包会把它裁掉表现为作业启动报ClassNotFoundException: RocksDBKeyedStateBackend。3.3 启动顺序与首次冒烟配置改完不要一把梭全启按 JobManager → HistoryServer → TaskManager 的顺序来每步看角色日志。JobManager 起来后日志里应该有Starting JobManager和 REST 端口监听TaskManager 起来后要在 JobManager 日志里看到Registered TaskManager这一步没出现说明网络或jobmanager.rpc.address配错了。# 冒烟提交 Flink 自带的 WordCount 示例jar 在 parcel 的 examples 下 flink run -c org.apache.flink.streaming.examples.wordcount.WordCount \ /opt/cloudera/parcels/FLINK/lib/flink/examples/streaming/WordCount.jar # 看作业列表确认提交成功 flink list第一段命令里的-c指定入口类路径按实际 parcel 解压位置调整CM 托管的 parcel 一般在/opt/cloudera/parcels/FLINK/。第二段flink list能列出 RUNNING 作业就说明整条链路通了。如果flink run卡在「Job has been submitted」但flink list看不到多半是 TaskManager 没注册上回去看 TaskManager 日志里的连接报错。4. 避坑与排查五个真实翻车现场4.1 现象parcel 激活成功但「添加服务」里没有 Flink原因通常是 csd jar 没被 CM 加载。CM 只在启动时扫描 csd 目录热放进去的 jar 不会立即生效。解决把 csd jar 放到/opt/cloudera/csd/权限cloudera-scm:cloudera-scm且 644然后重启 cloudera-scm-server。重启后还看不到检查 jar 内descriptor/service.sdl的name是否与 parcel 的name对应对不上 CM 会静默忽略。4.2 现象TaskManager 反复重启日志报内存不足原因多半是taskmanager.memory.process.size设得比容器实际可用内存大或者taskmanager.memory.managed.size与taskmanager.memory.task.heap.size之和超过了 process size。Flink 1.16 的内存模型是「总进程内存 堆 堆外 元空间 网络缓冲 托管内存」任何一项超配都会在启动时被校验拦下。解决先用taskmanager.memory.process.size定总盘子其余项留空让 Flink 按比例自动推导别手动逐项填死。4.3 现象作业能跑但 checkpoint 一直失败先看失败原因字段常见三种HDFS 权限不足、state.checkpoints.dir路径不存在、checkpoint 超时。权限问题给运行用户加目录写权限即可路径不存在要手动hdfs dfs -mkdir -p超时则调大execution.checkpointing.timeout同时看是不是状态太大导致单次快照时间过长必要时开增量 checkpointrocksdb 后端支持。这里有个后悔药checkpoint 配置改完要重启作业才生效别在运行中的作业上改完就等。4.4 现象Flink SQL 提交报 Hive 元数据连接失败CDH 集成场景下 Flink SQL 常要连 Hive Metastore报错一般是Could not connect to MetaStore或版本不匹配。原因是 Flink 1.16 自带的 Hive connector 版本与 CDH 6.3.2 的 Hive 2.1.1 不完全对齐需要把集群里的hive-exec、hive-metastore相关 jar 按官方文档的版本对照表替换到 Flink 的 lib 下。解决确认hive.version配置为 2.1.1并把 CDH 的 hive 配置目录通过HIVE_CONF_DIR指过去别用 Flink 包内自带的默认配置。4.5 现象Kerberos 环境下作业提交报认证失败CDH 集群开了 Kerberos 后Flink 需要有效的 keytab 和 principal。报错通常是GSS initiate failed或No valid credentials。原因有两个一是 CM 里 Flink 服务的 Kerberos principal 没配或 keytab 路径不可读二是提交作业的客户端机器没做kinit。解决在 CM 的 Flink 服务配置里填好 principal 和 keytab确保运行用户可读客户端提交前先kinit -kt拿票据票据过期时间要覆盖作业运行周期长跑作业建议配 delegation token 续期。5. 进阶把 Flink on CDH 跑稳的几个习惯部署通了只是开始真正决定这套集成能不能长期用的是几个日常习惯。第一个是版本锁定parcel 的depends区间、Flink 的hive.version、CDH 的 Hadoop 版本三者要记在一张表里任何一方升级前先对照我见过太多「顺手升了个小版本Flink SQL 全挂」的案例。第二个是资源隔离CM 里给 Flink 的 TaskManager 配 cgroup 或 YARN 队列别和 Hive、Spark 抢同一批机器的内存实时作业被批处理挤掉是最难查的一类问题。第三个习惯是监控落地。Flink 自带 metrics reporter接进 CM 的监控体系最省事的方式是配metrics.reporter.prom.class走 Prometheus再让 CM 的告警规则读同一份指标至少要盯住numRunningJobs、taskSlotsAvailable、lastCheckpointDuration三个数前两个反映容量后一个反映状态健康度。第四个是升级演练parcel 升级不是替换文件那么简单要先在测试集群走一遍「停作业 → 保存 savepoint → 换 parcel → 从 savepoint 恢复」savepoint 路径和兼容性确认无误再动生产。# 升级前保存 savepoint路径写 HDFS flink savepoint jobId hdfs:///flink/savepoints/ # 升级后从 savepoint 恢复 flink run -s hdfs:///flink/savepoints/savepoint-xxxx \ -c com.example.MyJob /path/to/job.jar第一段命令的jobId从flink list拿savepoint 目录要提前建好并给权限。第二段-s指定 savepoint 路径注意 Flink 1.16 对 savepoint 的兼容性有版本约束跨大版本恢复前查一下官方兼容表别想当然。我自己的习惯是每次动 parcel 之前先把当前所有 RUNNING 作业的 savepoint 打一遍并记下路径哪怕这次升级看起来「只是换个包」。从那以后我每次升级 Flink parcel 都强制走一遍 savepoint 演练宁可多花半小时也不想在凌晨对着恢复失败的作业干瞪眼。希望帮到你。本文还有配套的精品资源点击获取