CDH6.3.2上部署Flink1.16.2的依赖适配与生产调优指南

简介：CDH6.3.2是常见的大数据发行版，Flink-1.16.2可补充集群实时计算能力。针对CDH6.3.2环境集成Flink-1.16.2的现实需求，资源省去了自行制作parcel、csd以及适配Hadoop组件的繁琐过程。面向大数据平台运维与开发人员，尤其是通过Cloudera Manager管理集群、希望以parcel方式快速部署Flink的团队。压缩包共7个文件，大小约518MB，包含parcel安装文件、sha完整性校验文件、json元数据定义，以及命令行工具、YARN适配、Hadoop shaded依赖等jar组件，能够支撑Flink on YARN运行并补齐Hadoop依赖。包内已按CDH6.3.2常见部署方式配齐组件，parcel基于el7构建，可直接用于CentOS/RHEL 7系节点；csd服务描述文件与元数据定义配合，可被Cloudera Manager正确识别，安装后能在界面中管理Flink服务。文件分类明确，无需额外寻找依赖，适合内网离线环境使用。整体使用zip格式打包，便于传输与离线保存。对已有Flink使用经验、但不熟悉CDH集成细节的读者，该资源可帮助减少版本兼容性与打包排错成本。目前已有972人学习查看，适合作为CDH6.3.2上启用Flink-1.16.2的现成素材。

1. 在 CDH6.3.2 上部署 Flink 1.16.2：开局就是版本错位

Flink 1.16.2 搭配 CDH6.3.2，乍看只是一个常规的实时计算组件上集群，实际动手才会发现这是个典型的“版本错位”场景：CDH6.3.2 自带的 Hadoop 是 3.0.0-cdh6.3.2，而 Flink 1.16.2 官方默认自带的 Hadoop 客户端是 2.10.2，两者在依赖、类加载、RPC 协议上都有差异，直接部署大概率会翻车在 classpath 冲突上。这篇文章把这条路径拆开：从组件版本对应关系、依赖适配思路、最小可运行部署、踩坑记录到生产参数调整，覆盖你从“先试试”到“能上线”的完整过程。适合正在 CDH6.3.2 集群上做实时数仓、实时指标或者 Flink SQL 开发的同学，也适合想从 Standalone 模式迁到 YARN 上跑作业的团队参考。

2. 先别急着装：CDH6.3.2 和 Flink 1.16.2 的版本对应与依赖冲突清单

2.1 CDH6.3.2 集群自带的组件版本，先盘一遍

CDH6.3.2 虽然是 2020 年左右的发行版，但至今仍有很多生产集群在用。它的组件版本序列相对固定：Hadoop 是 3.0.0-cdh6.3.2，Hive 是 2.1.1-cdh6.3.2，HBase 是 2.1.0-cdh6.3.2，ZooKeeper 是 3.4.5-cdh6.3.2，Kafka 是 2.2.1-cdh6.3.2（如果用 CM 管理的话）。

这些版本里最影响 Flink 部署的是 Hadoop 和 Hive。Flink 1.16.2 官方发布包默认捆绑的是 hadoop-2.10.2 的客户端 jar，你要把它塞进一个 Hadoop 3.0.0 的集群里，第一个遇到的就是 javax.servlet、guava、commons-logging 等基础库版本冲突，第二个是 HDFS 客户端的 RPC 调用兼容性。第三个是如果你要读 Hive 表，hive-exec 的版本也得跟 CDH 的 2.1.1 对齐，否则连 metastore 都会连不上。

所以在摸清楚这些版本之前，别急着下载 Flink 包。我一般先把集群里各组件版本记下来，写成一张对应表，后续所有依赖排除都以这张表为准。

这张表不用背，但建议在部署文档里留一份。后边所有排错，回到这张表去查版本对应，能省很多时间。

2.2 Flink 1.16.2 到底依赖什么，官方包和源码包的区别

Flink 1.16.2 发布时提供了两种主要的二进制包：一种是不带 Hadoop 依赖的纯 Flink 包，另一种是带 hadoop-2.10.2 的包。很多人在官网下载页面会犹豫选哪个，这里直接给结论：部署到 CDH6.3.2 上，两个都不直接能用，需要手动调整依赖。

常见做法是下载不带 Hadoop 依赖的发布包，然后把 CDH 集群里的 hadoop-client、hadoop-hdfs-client、hadoop-yarn-client 等 jar 整理出来，放到 Flink 的 lib 目录下，或者通过HADOOP_CLASSPATH环境变量注入。第二种做法是下载带 Hadoop 2.10.2 的包，再排除掉自带 jar，换成 CDH 的。我建议用第一种，干净，后续升级 Flink 版本时不用重复处理 bundled 依赖。

还有一个细节是 flink-shaded-hadoop-uber 这个包，Flink 官方曾经提供过把 Hadoop 2.x 依赖打成一个大 jar 的版本，但这个 uber 包本身和 CDH 的 Hadoop 3.0 冲突很严重，尤其是 protobuf 和 guava。如果你在网上搜到老教程让你直接放 uber jar，千万别在 CDH6.3.2 上照做。正确的思路是：用 CDH 自己提供的 Hadoop 客户端 jar 集合，让 Flink 进程在运行时加载这些 jar，而不是反过来让 Hadoop 去适配 Flink 自带的旧版本。

2.3 适配方案怎么选：本地 classpath、HADOOP_CLASSPATH 还是改 lib 目录

我实际验证过三种方式，简单说下利弊。

第一种是修改flink-conf.yaml里的env.java.opts，把 Hadoop 相关的 classpath 用-classpath方式追加到 JVM 启动参数里。这种做法的好处是不污染 Flink 的 lib 目录，坏处是每个作业提交时都要带上这些参数，YARN application 模式下容易漏。

第二种是设置HADOOP_CLASSPATH环境变量，在提交作业的机器上执行export HADOOP_CLASSPATH=$(hadoop classpath)，Flink 提交脚本会自动读取这个变量。这个方式在 Standalone 模式下很好用，但在 YARN 模式下，ResourceManager 申请到的 Container 不一定继承这个环境变量，需要在flink-conf.yaml里把containerized.taskmanager.env.HADOOP_CLASSPATH也设置一遍，容易漏。

第三种是直接把需要的 jar 复制到 Flink 的 lib 目录下，同时把 Flink 自带冲突的 jar 删掉。这个方式最直观，也是我目前最推荐的。删 jar 的时候要小心，只删确定冲突的，比如老版本 guava、protobuf-java、hadoop 相关的 bundle 包，别把 flink 自身的核心 jar 误删了。

三种方式可以混用：lib 目录放 CDH 的 hadoop-client 相关 jar，flink-conf.yaml里加上HADOOP_CLASSPATH作为兜底，提交机器上也 export 一份。多一层保底，少一次踩坑。

2.4 需要的 jar 清单和从哪里拿

这里列一份我整理过的 jar 清单，来源是 CDH 集群中/opt/cloudera/parcels/CDH/lib/hadoop/client目录，这个目录下 CDH 已经帮你把运行 Hadoop 客户端所需的 jar 打包好了，比自己逐个找省事得多。

我一般会把这个目录下所有 jar 都复制过去，然后做减法。要特别注意的冲突点：CDH 自带的 guava 是 11.0.2，Flink 1.16.2 需要 guava 30 以上，这会直接导致NoSuchMethodError；CDH 自带的 protobuf-java 是 2.5.0，Hadoop 3.0 里已经升级到 2.5.0 以上，如果 Flink 任务里用到 protobuf 相关的序列化，必须把低版本排除掉。

实际操作时，我建议复制完 jar 后做一次全量启动测试，先跑一个最简单的 Flink 自带的 wordcount 或者 SQL 任务，看能不能正常提交和读取 HDFS。这一步能跑通，再继续接 Hive、Kafka 这些外部组件。不要一次把所有组件都接上，出了问题根本分不清是 Flink 自身问题还是 CDH 版本问题。

3. 动手部署：目录规划、配置文件与第一个 SQL 任务

3.1 下载 Flink 1.16.2 并规划目录结构

下载 flink-1.16.2-bin-scala_2.12.tgz 这个包，不需要 hadoop 版本的那一个。解压后我通常会按下面的目录规划来组织：

# 假设 Flink 安装根目录为 /data/flink mkdir -p /data/flink/flink-1.16.2 mkdir -p /data/flink/flink-1.16.2/lib/cdh # 存放 CDH 的 hadoop client jar mkdir -p /data/flink/flink-1.16.2/opt/connectors mkdir -p /data/flink/data/checkpoints mkdir -p /data/flink/data/savepoints

lib/cdh 这个目录用于集中存放从 CDH 集群拷贝出来的 jar，不直接丢到 lib 目录下，方便后面出问题的时候对比和排查。opt/connectors 目录放后续需要的 flink-sql-connector-kafka 等连接器，避免和核心 lib 混在一起。

把 Flink 解压后，第一件事不是改配置，而是检查自带的 lib 目录里有哪些 jar，记下来。后续删 jar 的时候，以这份清单为基准，不要动清单之外的任何东西。

3.2 从 CDH 客户端目录复制 Hadoop jar 到 Flink lib

这一步的关键命令是找到 CDH 的 hadoop client 目录。不同集群安装路径可能不一样，我这边是通过 Cloudera Manager 安装的，路径一般是/opt/cloudera/parcels/CDH/lib/hadoop/client，你可以用下面的命令确认：

# 在 CDH 集群的任意一台边缘节点上执行 cd /opt/cloudera/parcels/CDH/lib/hadoop/client ls -l *.jar | wc -l

正常情况下会输出 20 到 40 个 jar。把这些 jar 复制到 Flink 的 lib/cdh 目录下，然后选择性放一部分到 Flink 的 lib 目录。我的做法是把整个 client 目录的 jar 先全部复制到 lib/cdh，再从里面挑出必要的放到 lib 目录下：

cd /opt/cloudera/parcels/CDH/lib/hadoop/client cp *.jar /data/flink/flink-1.16.2/lib/cdh/ cd /data/flink/flink-1.16.2/lib/cdh # 只复制必需的 jar 到 lib 目录，guava 先不放进去，后面专门处理 cp hadoop-common-*.jar hadoop-hdfs-*.jar hadoop-client-*.jar \ hadoop-yarn-*.jar hadoop-mapreduce-client-core-*.jar \ commons-configuration2-*.jar commons-lang3-*.jar \ /data/flink/flink-1.16.2/lib/

这里有个关键点：不要一股脑把 client 目录下所有 jar 都复制进 lib。CDH 的 client 目录里有些 jar 和 Flink 自身带的重名，比如log4j、slf4j，复制进去后会出现重复类定义，启动时的报错会非常隐蔽。宁可先少放几个，跑一个最简单的 HDFS 读写测试，再逐步增加。

复制完后，检查 Flink 的 lib 目录里是否同时存在 Flink 自带的 hadoop 相关 jar，比如flink-shaded-hadoop-2-uber-*.jar，如果有，先删掉，留着必出问题：

ls /data/flink/flink-1.16.2/lib/ | grep hadoop # 如果看到 flink-shaded-hadoop-2-uber-*.jar # rm /data/flink/flink-1.16.2/lib/flink-shaded-hadoop-2-uber-*.jar

删除后，再检查是否有低版本 guava 和 protobuf，这些后续在第 4 章避坑部分细说。

3.3 修改 flink-conf.yaml 的核心参数

flink-conf.yaml是 Flink 所有运行时配置的入口，CDH6.3.2 场景下我通常重点改这几个参数，先给一份能跑起来的初始配置：

# 路径配置 jobmanager.memory.process.size: 2048m taskmanager.memory.process.size: 4096m taskmanager.numberOfTaskSlots: 2 parallelism.default: 2 # 高可用配置，使用 ZooKeeper high-availability: zookeeper high-availability.zookeeper.quorum: zk1.example.com:2181,zk2.example.com:2181,zk3.example.com:2181 high-availability.storageDir: hdfs:///flink/ha high-availability.zookeeper.path.root: /flink # 状态后端和 Checkpoint state.backend: hashmap state.checkpoints.dir: hdfs:///flink/checkpoints state.savepoints.dir: hdfs:///flink/savepoints # 允许作业恢复 execution.checkpointing.interval: 60s execution.checkpointing.min-pause: 30s execution.checkpointing.tolerable-failed-checkpoints: 3 # 通信和 Web UI rest.port: 8081 jobmanager.rpc.address: localhost jobmanager.rpc.port: 6123

解释一下几个关键参数。high-availability: zookeeper是让 Flink 的 JobManager 在异常退出后能自动恢复，CDH6.3.2 自带的 ZooKeeper 3.4.5 支持 Flink 1.16.2 的 HA 协议，但注意 root path 不要和集群里其他组件的 znode 冲突，建议用独立的/flink前缀。state.backend: hashmap是让 Flink 用堆内存保存状态，适合先跑通功能，生产环境一般换 rocksdb，后面第 6 章会说。

jobmanager.rpc.address: localhost这一项在单机测试时可以这样写，但如果是多节点部署，必须改成实际运行 JobManager 的节点 IP 或域名。CDH 集群上我一般会把 JobManager 单独部署在一台节点上，和集群管理节点分开，避免资源争抢。

3.4 配置 core-site.xml 和 hdfs-site.xml

Flink 访问 HDFS 时，需要读取 CDH 集群的core-site.xml和hdfs-site.xml，否则它会用默认的 HDFS 配置，连的可能是 localhost:8020 或者压根找不到 NameNode。

通常做法是从 CDH 集群任意节点上把这两个文件拷贝到 Flink 的 conf 目录下：

cp /etc/hadoop/conf/core-site.xml /data/flink/flink-1.16.2/conf/ cp /etc/hadoop/conf/hdfs-site.xml /data/flink/flink-1.16.2/conf/ cp /etc/hadoop/conf/yarn-site.xml /data/flink/flink-1.16.2/conf/ cp /etc/hadoop/conf/mapred-site.xml /data/flink/flink-1.16.2/conf/

如果你的 CDH 环境启用了 Kerberos，还需要把hive-site.xml也拷贝过来，并且后续要在 Flink 配置里加上 Kerberos 相关的参数。这一部分在避坑章节里单独展开，这里先不占篇幅。

拷贝完成后，在 Flink 的 conf 目录下执行一个简单命令验证能不能读到 HDFS：

cd /data/flink/flink-1.16.2 bin/hadoop fs -ls hdfs:/// # 注意：这个命令需要依赖 lib 目录下的 hadoop jar

如果报错ClassNotFoundException: org.apache.hadoop.conf.Configuration，说明 lib 目录下缺少 hadoop-common 的 jar，回去检查 3.2 步是否复制完整。如果报错Connection refused，说明hdfs-site.xml里的 NameNode 地址没被读到，检查 conf 目录下是否真的有这个文件。

3.5 启动 Flink 集群并提交第一个 SQL 任务

依赖配置好了，先做一次 Standalone 模式的启动，验证集群本身没问题。然后再尝试 YARN 模式。

cd /data/flink/flink-1.16.2 bin/start-cluster.sh # 查看日志，确认没有 ClassNotFound 或 NoSuchMethodError tail -f log/flink-*-standalonesession-*.log

正常启动后，访问http://jobmanager节点:8081，能看到 Flink Web UI，说明基础环境没问题。接下来提交一个最简单的 SQL 任务验证整体链路。

这里我建议用 Flink SQL Client，比写 Java 代码快很多，而且能直接看到 SQL 层面的报错信息。先创建一张 Source 表，直接读 HDFS 上的文本文件：

CREATE TABLE words ( word STRING ) WITH ( 'connector' = 'filesystem', 'path' = 'hdfs:///tmp/words.txt', 'format' = 'csv' ); SELECT word, COUNT(*) AS cnt FROM words GROUP BY word;

这个 SQL 任务如果能在 Web UI 上看到正常 RUNNING，并且输出结果，说明 Flink 1.16.2 和 CDH6.3.2 的 HDFS 连通性没问题。如果这里就报错，先别急着往下走，把报错信息里的ClassNotFoundException或者NoSuchMethodError记下来，逐一去 lib 目录排查冲突 jar。

4. 部署与联调中绕不开的 5 个坑：从 JAR 冲突到高可用配置

4.1 Guava 版本太低导致 HDFS 客户端直接 NoSuchMethodError

现象：启动 Flink 集群正常，但提交任务后，JobManager 日志里出现java.lang.NoSuchMethodError: com.google.common.base.Preconditions.checkArgument(ZLjava/lang/Object;)V或者com.google.common.util.concurrent.MoreExecutors.directExecutor()找不到。这通常是 CDH 自带的 guava 11.0.2 被加载到了类路径上，而 Flink 1.16.2 内部很多代码用的是 guava 30 的 API。

原因：CDH 的 hadoop-client 目录下自带 guava-11.0.2.jar，当我们把整个 client 目录的 jar 都复制到 Flink lib 后，这个低版本 guava 就被加载了。JVM 加载类时按 classpath 顺序，一旦先加载了 11.0.2，Flink 内部调用 guava 新版本 API 的地方就会全部崩掉。

解决：从 Flink lib 目录里删掉 CDH 自带的 guava 低版本 jar，同时确认 Flink 自身有没有带更高版本 guava。Flink 官方发布包的 lib 目录通常不直接提供 guava，而是通过 shaded 的方式把 guava 打包进了flink-dist相关 jar 里，所以直接删除低版本 guava 即可。操作如下：

cd /data/flink/flink-1.16.2/lib ls *guava*.jar rm -f guava-11.0.2.jar # 删除 CDH 自带的低版本

如果你后续发现某些 CDH 组件必须依赖 guava 11，不能直接删，那就换思路：把 guava 11 保留在 CDH 自己的 classpath 里，Flink 启动时不要加载它。最简单的办法是启动 Flink 时通过env.java.opts指定-Xbootclasspath/p:或者调整类加载顺序，但这个方案复杂度高，非必须的话不建议用。

4.2 Flink 自带的 shaded Hadoop jar 污染 classpath

现象：提交任务到 YARN 时，报错java.lang.VerifyError或者org.apache.hadoop.fs.FileSystem加载了多个不同版本。有时候甚至出现NoClassDefFoundError: org/apache/hadoop/conf/Configuration这种看起来不该缺的类。

原因：Flink 1.16.2 如果下载的是带 Hadoop 的发布包，lib 目录里会有flink-shaded-hadoop-2-uber-*.jar；即使下载的是不带 Hadoop 的包，某些连接器或者示例 jar 里也可能嵌入旧版 Hadoop 类。CDH6.3.2 是 Hadoop 3.0，两套 Hadoop 类同时存在必然冲突。

解决：一键清理 Flink lib 目录下的所有 shaded Hadoop jar：

cd /data/flink/flink-1.16.2/lib rm -f flink-shaded-hadoop-*.jar

清理后，重新确认只有 CDH 的 hadoop client jar 在 lib 目录里。判断标准是执行grep -L "cdh" *.jar应该没有任何输出与 hadoop 核心类相关。注意有些 Flink 自带的 jar 名称里没有 hadoop 字样，但内部嵌入了旧版类，这类 jar 需要逐个通过jar tf检查，通常在opt/目录下，不放进 lib 就不会有问题。

4.3 ZooKeeper 高可用配置了但 JobManager 无法恢复

现象：按照第 3 章的配置设置了high-availability: zookeeper，JobManager 进程异常 kill 后，Web UI 能重新起来，但提交的作业全部丢失，没有自动恢复。查看日志发现 Flink 连上了 ZooKeeper，但找不到存储作业信息的节点。

原因：CDH6.3.2 自带的 ZooKeeper 3.4.5 默认有zookeeper.sasl.client相关配置，如果集群开启了 Kerberos 或者设置了 ACL 权限，Flink 客户端没有对应的认证配置，它在 ZooKeeper 里创建不了 znode。另外，high-availability.storageDir: hdfs:///flink/ha这个目录如果没有在 HDFS 上提前创建，Flink 写入 HA 数据时会抛FileNotFoundException。

解决：确认两点。第一，HDFS 上的目录提前建好，命令是hdfs dfs -mkdir -p /flink/ha /flink/checkpoints /flink/savepoints，并且给运行 Flink 的用户加上读写权限。第二，如果 ZooKeeper 开启了 SASL 认证，需要在flink-conf.yaml里加上以下配置：

high-availability.zookeeper.client.acl: creator security.kerberos.login.contexts: Client security.kerberos.login.keytab: /etc/security/keytabs/flink.keytab security.kerberos.login.principal: flink/_HOST@EXAMPLE.COM

这里最容易忽略的是security.kerberos.login.contexts: Client，这个参数告诉 Flink 在建立 ZooKeeper 连接时使用 Kerberos 认证上下文。很多 CDH 集群默认开启了 ZooKeeper 的 SASL，如果没有这个参数，Flink 连 ZooKeeper 时会被判定为 unauthenticated，权限不足，HA 自然失效。

4.4 Hive 版本不匹配导致 HiveCatalog 初始化失败

现象：在 Flink SQL 里执行CREATE CATALOG myhive WITH ('type'='hive', 'hive-conf-dir'='/etc/hive/conf')，然后执行USE CATALOG myhive时报NoClassDefFoundError: org/apache/hadoop/hive/ql/metadata/HiveException，或者java.lang.NoSuchMethodError: org.apache.hadoop.hive.conf.HiveConf.getVar(Lorg/apache/hadoop/hive/conf/HiveConf$ConfVars;)Ljava/lang/String;。

原因：Flink 1.16.2 官方默认编译时依赖 Hive 2.3.x 或者 3.1.x，你放入 lib 目录的 hive-exec 是 CDH6.3.2 自带的 2.1.1-cdh6.3.2，API 有差异。特别是HiveConf.getVar()方法在 Hive 版本间签名变化比较频繁，版本对不上直接就 NoSuchMethodError。

解决：不要只用 Flink 官方提供的 hive connector 依赖，而是把 CDH 集群里的 hive-exec 和相关 hive 依赖 jar 全部放进 Flink lib 目录，替换掉 Flink 自带的版本。具体操作：

cd /opt/cloudera/parcels/CDH/lib/hive/lib cp hive-exec-*.jar /data/flink/flink-1.16.2/lib/ cp hive-metastore-*.jar /data/flink/flink-1.16.2/lib/ cp hive-service-*.jar /data/flink/flink-1.16.2/lib/ cp libfb303-*.jar /data/flink/flink-1.16.2/lib/

复制完以后，还要把 Flink 自带的旧版本 hive 相关 jar 删掉，否则两个版本的 hive-exec 在 classpath 里互相打架。删除时注意看清楚，只删flink-sql-connector-hive-*或hive-exec-2.3.*这类官方包自带的，别把 CDH 自己刚复制进去的删了。

另外一个隐藏点：CDH 的 hive-exec jar 依赖calcite和derby，这些库的版本在 Flink lib 里可能被覆盖掉。如果初始化 hive catalog 时出现NoSuchMethodError: org.apache.calcite...，需要从 CDH 的 hive/lib 目录下把对应的 calcite jar 也复制一份进 Flink lib，让 Hive 相关的类统一走 CDH 的版本。

4.5 Kerberos 环境下 keytab 配置位置不对，作业可以启动但读不了 HDFS

现象：Flink 作业能在 YARN 上跑起来，但每当要访问 HDFS 时，报javax.security.auth.login.LoginException: Unable to obtain password from user或者org.apache.hadoop.security.AccessControlException: SIMPLE authentication is not enabled。更诡异的是，在提交作业的机器上手动执行hadoop fs -ls是正常的，但 Flink 任务一跑就挂。

原因：Flink 的 YARN 容器默认不会加载提交机器上的 Kerberos ticket。如果在flink-conf.yaml里没有显式配置 keytab，作业提交后 Container 里的 JVM 拿不到 Hadoop 的 delegation token，访问 HDFS 时自然没有认证信息。CDH6.3.2 开启 Kerberos 后，这个坑出现的概率很大。

解决：把 keytab 和 principal 配置写进flink-conf.yaml，不要只依赖外部kinit：

security.kerberos.login.use-ticket-cache: false security.kerberos.login.keytab: /etc/security/keytabs/flink.keytab security.kerberos.login.principal: flink@EXAMPLE.COM

如果提交的是 SQL 任务，还要确保flink-sql-client启动时也能读到这个配置。我遇到过一个情况：keytab 路径写的是相对路径./flink.keytab，JobManager 在 YARN 上启动时工作目录在容器内部，相对路径根本找不到。所以 keytab 一定要写绝对路径，并且保证 YARN 容器所在的每台节点上都有这个文件，或者把 keytab 放在 HDFS 上通过security.kerberos.login.keytab: hdfs:///etc/flink.keytab这种方式引用。

5. 验证一套可复现的 Demo：SQL Client 跑通实时统计

5.1 先构造一个最小数据集，写入 HDFS 和 Kafka

验证 Flink 1.16.2 与 CDH6.3.2 的兼容性，最好的方式是模拟一条完整的实时链路：数据源到 Kafka，Flink SQL 消费 Kafka，处理后写入 HDFS，最后再用批式查询读回来。

我先在 HDFS 上创建测试目录，并放一个简单的单词文本文件：

hdfs dfs -mkdir -p /tmp/flink-demo/input echo -e "flink\ncdh\nflink" > /tmp/words.txt hdfs dfs -put /tmp/words.txt /tmp/flink-demo/input/

接着往 Kafka 里写入几条实时数据。如果 CDH6.3.2 集群里有 Kafka，直接用kafka-console-producer写入：

kafka-topics --create --topic flink-demo-topic --partitions 3 --replication-factor 1 \ --zookeeper zk1.example.com:2181 kafka-console-producer --broker-list kafka1.example.com:9092 --topic flink-demo-topic # 输入： # {"word": "flink", "cnt": 1} # {"word": "cdh", "cnt": 1} # {"word": "flink", "cnt": 1}

没有 Kafka 也可以先用 filesystem connector 验证，但建议生产环境评估还是把 Kafka 加进来，因为 Kafka 到 Flink SQL 的 connector 在 CDH 环境里也有一些依赖坑，比如普通kafka-clients版本和flink-sql-connector-kafka内置版本不一致，早验证早安心。

5.2 启动 SQL Client 并绑定 HiveCatalog

用 SQL Client 的好处是能直接验证 Flink 对 CDH HiveCatalog 的支持。启动前先用命令行验证 Hive 配置能读到：

cd /data/flink/flink-1.16.2 bin/sql-client.sh embedded -d conf/sql-client.yaml

sql-client.yaml里可以预置 catalog 配置：

catalogs: - name: cdh_hive type: hive hive-conf-dir: /etc/hive/conf default-database: default

进入 SQL Client 后，执行以下语句验证 Catalog 是否连通：

USE CATALOG cdh_hive; SHOW DATABASES;

如果能列出数据库列表，说明 Flink 1.16.2 访问 CDH Hive Metastore 没问题。如果不能，看一下报错里有没有 Hive 版本相关字样，回到 4.4 的解决思路去排查 jar。

5.3 创建 Kafka Source 表和 HDFS Sink 表

Catalog 连通后，创建两张核心表。第一张读 Kafka：

CREATE TABLE kafka_words ( word STRING, cnt INT, ts TIMESTAMP(3) METADATA FROM 'timestamp' ) WITH ( 'connector' = 'kafka', 'topic' = 'flink-demo-topic', 'properties.bootstrap.servers' = 'kafka1.example.com:9092,kafka2.example.com:9092', 'properties.group.id' = 'flink-demo-group', 'scan.startup.mode' = 'earliest-offset', 'format' = 'json' );

注意ts TIMESTAMP(3) METADATA FROM 'timestamp'，这个语法是 Flink 1.16.x 的标准写法，低版本用的是proctime或者rowtime，不要混淆。scan.startup.mode设为earliest-offset是为了确保测试时 Kafka 里已有数据能被全部消费到。

第二张表写 HDFS，以 Hive 分区表为例：

CREATE TABLE hive_word_stats ( word STRING, total_cnt BIGINT, stat_date STRING ) PARTITIONED BY (stat_date) WITH ( 'connector' = 'hive', 'default-database' = 'default', 'partition.time-extractor.timestamp-pattern' = '$stat_date' );

这里用 Hive 表作为 Sink，实际上也是验证 Flink 写 Hive 的 ACID 能力和分区发现机制。CDH6.3.2 的 Hive 2.1.1 对 Orc 格式支持比较成熟，如果你用的表格式是 Parquet，也可以把'format'指定为'parquet'，但 ORC 在 CDH 环境里测试通过率更高。

5.4 运行统计任务并检查结果

执行下面的插入语句，把 Kafka 里的数据做聚合后写入 Hive 表：

INSERT INTO hive_word_stats SELECT word, SUM(cnt) AS total_cnt, DATE_FORMAT(ts, 'yyyy-MM-dd') AS stat_date FROM kafka_words GROUP BY word, DATE_FORMAT(ts, 'yyyy-MM-dd');

这个任务提交后，观察 Flink Web UI 上的 Job 状态。正常情况下会先看到RUNNING，然后每隔几十秒出现一次Checkpoint完成记录。等到任务运行一段时间后，切回 SQL Client 执行查询：

SELECT * FROM hive_word_stats;

如果查询结果有数据，说明整条链路全部打通。这里有个细节要说明：Hive 表默认是分区表，Flink 写入数据后，如果查询没有指定分区，可能查不到刚写入的数据。可以先用SHOW PARTITIONS hive_word_stats确认分区是否存在。CDH 的 HiveServer2 对 Hive 分区有缓存，刚写入的分区偶尔需要执行一次MSCK REPAIR TABLE hive_word_stats才能及时被发现。

5.5 用日志和 Web UI 确认没有隐藏的类冲突

不要以为任务能跑通就万事大吉。有些类冲突发生在任务启动初期，但被 Flink 的容错机制吞掉了，后续运行中一旦走到特定逻辑，就会突然抛异常。我的习惯是跑一个任务后，主动查看三类日志：

第一类是 JobManager 日志，路径在log/flink-*-standalonesession-*.log，重点搜ERROR和Exception，把堆栈中的类名和 lib 目录下的 jar 做对照。第二类是 TaskManager 日志，同样的关键字，但这里的异常往往更接近业务代码。第三类是 Flink Web UI 上 Job 的Exceptions标签页，这里能看到所有子任务的失败原因。

我遇到过一个情况：任务运行 20 分钟后突然失败，日志里只有一行java.lang.NoClassDefFoundError: org/apache/hadoop/mapred/JobConf，而这个类明显在 hadoop-mapreduce-client-core jar 里。原因是任务里使用了 Hive 的某种输出格式，触发了 MapReduce 相关的类加载。这种情况只有通过查看运行期日志才能发现，前期 SQL 校验阶段完全看不出来。

6. 从 Demo 到生产：几个值得改的默认参数和后续动作

6.1 内存模型别只盯着堆内存，三块内存都要配

Flink 1.16.2 的 TaskManager 内存模型分堆内、托管内存和直接内存三块。默认配置下，taskmanager.memory.process.size只定义总大小，框架会按默认比例划分。但在 CDH6.3.2 上跑真实作业时，默认比例经常出问题：如果状态后端用 RocksDB，托管内存占比默认只有 40%，不够用；如果网络缓冲和 RPC 通信频繁，直接内存占比也可能不足。

我一般会显式设置这三块的比例，不依赖默认值：

taskmanager.memory.process.size: 8192m taskmanager.memory.framework.heap.size: 256m taskmanager.memory.framework.off-heap.size: 32m taskmanager.memory.task.heap.size: 4096m taskmanager.memory.managed.size: 3072m taskmanager.memory.jvm-overhead.size: 1024m

这里task.heap.size给业务逻辑用，managed.size给状态后端和 RocksDB 用，jvm-overhead给 JVM 自身和 native 库用。加起来要小于process.size，否则启动会报内存配置不合法。CDH 集群上单节点资源有限，taskmanager.numberOfTaskSlots我一般控制在 4 到 8 之间，避免 Slot 过多造成频繁 GC。

6.2 Checkpoint 策略：从 60 秒间隔到 10 分钟一次的稳定节奏

开发环境默认的 checkpoint 间隔 60 秒是方便调试的，生产环境这个值太激进。我见过不止一次生产事故：Checkpoint 间隔太短，状态很大时每次 Checkpoint 都要把全部状态写入 HDFS，YARN 容器被频繁网络 I/O 拖垮，反而导致作业不断失败重启。

生产环境我建议这样调：

execution.checkpointing.interval: 10min execution.checkpointing.min-pause: 5min execution.checkpointing.tolerable-failed-checkpoints: 3 execution.checkpointing.timeout: 10min state.backend.incremental: true

min-pause大于interval的 50%，这是避免两个 Checkpoint 同时进行的硬性要求。timeout设置为interval的 1 倍，避免单个 Checkpoint 卡死。state.backend.incremental: true是 RocksDB 增量 Checkpoint 的关键开关，只对 RocksDB 生效，能大幅减少每次 Checkpoint 写入的数据量。

另外一个 CDH 环境特有的问题：HDFS NameNode 如果开启了 HA，Checkpoint 写入时的 retry 策略要调一下。默认的fs.hdfs.temp.dir权限如果不够，写入失败后会进入无限重试。建议在flink-conf.yaml里加上：

env.java.opts: "-Dfs.hdfs.temp.dir=/tmp -Ddfs.client.block.write.replace-datanode-on-failure.policy=NEVER"

第一个参数确保 HDFS 临时目录可写，第二个参数是 Hadoop 3.x 推荐的 DataNode 写入失败策略调整，能减少部分网络抖动场景下的 Checkpoint 失败概率。

6.3 状态后端选型：生产优先 RocksDB，别用 HashMap

开发环境我用hashmap状态后端，纯内存速度快，但生产作业状态超过几百 MB 后就开始吃紧。Flink 1.16.2 里切换 RocksDB 很简单：

state.backend: rocksdb state.backend.rocksdb.memory.managed: true state.backend.rocksdb.block.cache.size: 512m state.backend.rocksdb.writebuffer.count: 4 state.backend.rocksdb.writebuffer.size: 64m

memory.managed: true让 RocksDB 使用托管内存，由 Flink 统一管理，避免 RocksDB 的堆外内存和 JVM 堆内存互相抢占。writebuffer.size调大后，写入吞吐会明显提升，但每个 Slot 都会有一份独立的 writebuffer，Slot 数乘以 writebuffer 大小就是总开销，分配时要算清楚。

切换 RocksDB 后一定要重跑一遍完整 Demo，因为 RocksDB 的本地存储会写入 TaskManager 的临时目录。如果临时目录所在的磁盘容量不够，作业启动时会直接报磁盘空间不足。CDH 集群上我习惯把taskmanager.env.java.opts里加一句-Djava.io.tmpdir=/data/flink/tmp，确保 RocksDB 的临时文件落在数据盘而不是系统盘。

6.4 部署模式建议：直接上 YARN Application 模式

开发调试用 Standalone 模式没问题，但生产环境建议直接用 YARN Application 模式，让 Flink 作业作为 YARN 上的一个 Application 运行。CDH6.3.2 的 YARN 是 Hadoop 3.0，对 Application 模式支持很完整，资源隔离、日志聚合、失败重启都靠 YARN 管，比 Flink 自己管理省心得多。

提交命令一行：

cd /data/flink/flink-1.16.2 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-prod \ -c your.main.Class /path/to/your-job.jar

这里特别提醒一个点：-t yarn-application模式下，JobManager 和 TaskManager 都运行在 YARN 容器里，flink-conf.yaml里的jobmanager.rpc.address配置会被自动覆盖为容器的实际地址，不需要手动设置。如果你在配置里写死了内网 IP，反而会导致容器启动后无法互相通信。

6.5 从一次真实翻车里学到的：上线前先做三天稳定性验证

最后分享一个我自己的习惯。不管新集群还是新组件版本，上线前我都会设计一个“三天验证计划”：第一天跑一个简单的消费 Kafka 写 HDFS 的作业，持续 8 小时，观察 Checkpoint 是否稳定；第二天跑一个带状态聚合的作业，模拟业务峰值的双倍数据量，观察 GC 频率和背压情况；第三天做故障演练，手动 kill 掉 TaskManager，观察作业能否自动恢复，检查数据是否有重复或丢失。

这个习惯帮我在 CDH6.3.2 和 Flink 1.16.2 的组合上提前发现了不少问题，比如 HDFS 吞吐瓶颈、RocksDB 磁盘占用、ZooKeeper 连接数限制等。这些问题如果直接在生产上线后爆发，定位成本是验证阶段的十倍不止。希望这篇笔记能让你在这条路上少走几个弯路，希望帮到你。

本文还有配套的精品资源，点击获取