Flink CDC 写入 Kafka 后字段为 null 的排查与解决

1. 问题背景与现象拆解Flink CDC 这套链路在实时数仓里已经算是标配了MySQL binlog 抓过来Flink 做一层轻量 ETL再往 Kafka 里一丢下游不管是实时大屏、风控引擎还是离线归档都能接。链路本身不复杂但真正在生产上跑起来坑一点都不少。我最近就碰到一个特别典型的问题Flink CDC 把数据写进 Kafka 之后下游消费出来的 JSON 里一堆字段是 null而且不是偶发是稳定复现某些字段从头到尾就没见过值。这个现象乍一看很吓人因为第一反应往往是数据丢了。但实际排查下来你会发现绝大多数 null 问题都不是 CDC 抓取阶段丢的而是序列化阶段、Schema 映射阶段、或者 Kafka 消息体构造阶段出的问题。数据其实好好地躺在 Flink 的 RowData 里只是最后一步没被正确翻译成 JSON 字段。这篇文章我打算把整个排查过程完整复盘一遍从现象特征、可能原因、逐层定位到最终解决方案和避坑经验全部摊开讲。适合正在用 Flink CDC Kafka 做实时同步的同学尤其是那些刚把链路跑通、结果发现下游字段对不上的朋友。看完你至少能做到两件事第一遇到 null 不再瞎猜第二知道该从哪一层开始切。先说清楚这套链路的典型形态方便后面定位。常见写法是MySQL CDC Source - 一些转换算子 - Kafka Sink中间可能用 Flink SQL 写也可能用 DataStream API 写。两种写法出 null 的根因不完全一样但排查思路是相通的。下面我按先定位、再解释、后解决的顺序展开。2. 先搞清楚 null 到底出在哪一层2.1 三种 null 的本质区别很多人一看到 null 就慌了其实要先分清是哪种 null。我把它分成三类字段级 nullJSON 里字段存在但值是 null比如{id:1,name:null}。整条 nullKafka 消息体直接是null字符串或者整个 value 为空。字段缺失JSON 里压根没这个 key下游解析时默认当成 null。这三种的根因完全不同。字段级 null 通常是序列化器没拿到值或者Schema 对不上整条 null 往往是反序列化失败或者转换算子返回了 null字段缺失则多半是Schema 定义里没包含这个字段或者JSON 序列化时把 null 字段过滤掉了。我这次遇到的是第一种和第三种混在一起一部分字段是显式 null一部分字段直接消失。这就说明问题不在 CDC 抓取而在Schema 到 JSON 的映射环节。2.2 用最小化手段快速定位定位这件事最忌讳一上来就改代码。我的习惯是先加一个观测点把数据在进入 Kafka Sink 之前打印出来。如果是 Flink SQL可以直接把结果表改成 print sink 跑一遍CREATE TABLE print_sink ( id BIGINT, name STRING, amount DECIMAL(10,2) ) WITH (connector print); INSERT INTO print_sink SELECT * FROM mysql_cdc_source;如果 print 出来的数据字段是完整的那问题就在 Kafka Sink 的序列化配置上如果 print 出来就已经是 null那问题在更上游可能是 CDC Source 的 Schema 推断或者转换算子。这一步看着笨但能帮你省掉至少一半的排查时间。我见过太多人直接盯着 Kafka 里的消息看结果绕了一大圈才发现是上游 SQL 写错了。提示print sink 会把数据打到 TaskManager 的日志里本地跑没问题生产环境慎用数据量大的时候日志会被刷爆。2.3 一个容易被忽略的观测点Kafka 消息的 key除了 valueKafka 消息的 key 也经常出问题。如果你的 Sink 配置里指定了主键作为 key但主键字段在转换过程中变成了 null那 key 就会是空。下游如果按 key 做分区或者去重就会出现数据倾斜或者丢数据。所以排查的时候key 和 value 都要看。3. 根因一Schema 映射错位导致字段对不上3.1 Flink SQL 里字段顺序的坑这是最常见、也最隐蔽的一个坑。Flink SQL 在定义结果表的时候字段顺序必须和 SELECT 出来的字段顺序严格一致。如果你写的是CREATE TABLE kafka_sink ( name STRING, id BIGINT, amount DECIMAL(10,2) ) WITH (connector kafka, ...); INSERT INTO kafka_sink SELECT id, name, amount FROM source;看起来没问题对吧但 Flink 是按位置匹配的不是按名字。结果就是id的值被塞进了namename的值被塞进了id类型对不上的地方直接变成 null。这种错误在字段少的时候还能靠肉眼发现字段一多比如二三十个字段基本就是灾难。我这次踩的就是这个坑的变种源表字段顺序和结果表字段顺序不一致中间还夹着几个计算字段导致后面所有字段整体错位类型不匹配的位置全部变成 null。3.2 用显式列名规避顺序问题解决办法很简单永远显式写列名不要依赖位置匹配INSERT INTO kafka_sink (id, name, amount) SELECT id, name, amount FROM source;这样即使两边字段顺序不一样Flink 也会按名字去匹配。代价是写起来啰嗦一点但能救命。我的建议是只要字段超过 5 个就老老实实写列名别图省事。3.3 CDC Source 的 Schema 推断问题还有一种情况是 CDC Source 本身的 Schema 推断出了问题。MySQL CDC 连接器在启动时会去读表结构如果表结构在运行期间发生了变更比如加字段、改类型而 Flink 作业没有重启就会出现 Schema 不一致。这时候新字段读不到旧字段类型对不上表现就是一堆 null。这种情况的典型特征是作业刚启动时正常跑了一段时间后开始出 null。如果你观察到这个规律基本可以锁定是 Schema 演进的问题。解决办法有两个一是开启 Flink 的 Schema 演进支持scan.incremental.schema-change.enabled二是变更表结构后重启作业。4. 根因二序列化器配置不当4.1 JSON 序列化器的默认行为Flink 写 Kafka 最常用的格式是json底层用的是 Jackson。这里有个很多人不知道的默认行为Jackson 默认会把值为 null 的字段也序列化出来也就是{name:null}。但如果你用的是某些自定义序列化器或者配置了NON_NULL策略null 字段就会被直接过滤掉表现就是字段缺失。这两种行为本身都没错问题在于下游的预期。如果下游按固定 Schema 解析字段缺失就会报错或者补 null如果下游是动态解析那字段缺失和显式 null 的语义是不一样的。所以第一步要确认你的序列化器到底是怎么处理 null 的。4.2 自定义序列化器的常见错误用 DataStream API 的时候很多人会自己写一个KafkaSerializationSchema。我见过最多的错误是在serialize方法里直接对 RowData 做getField但没做类型判断public byte[] serialize(RowData element) { JSONObject obj new JSONObject(); obj.put(id, element.getLong(0)); obj.put(name, element.getString(1)); return obj.toString().getBytes(StandardCharsets.UTF_8); }这段代码看着没问题但如果element.getString(1)返回的是 null某些 JSON 库会直接抛异常或者把整个对象序列化成 null。更隐蔽的是如果字段索引写错了比如getString(1)实际取的是别的字段类型不匹配时也会返回 null。我的建议是自定义序列化器里每个字段都要做 null 判断和类型校验宁可多写几行也别让 null 悄悄溜过去。4.3 Debezium 格式的嵌套结构如果你用的是debezium-json格式那要特别注意它的嵌套结构。Debezium 格式的消息体长这样{ before: {...}, after: {...}, op: c, ts_ms: 1234567890 }真正的数据在after里面。如果你在 Sink 里直接按平铺字段去解析那所有字段都会是 null因为根节点根本没有这些字段。这个坑我踩过一次排查了半天才发现是格式理解错了。5. 根因三转换算子里的 null 传播5.1 UDF 返回 null 的连锁反应Flink SQL 里用 UDF 做字段转换是很常见的操作比如格式化时间、拼接字符串、做金额单位换算。但如果 UDF 在某些输入下返回了 null这个 null 会一路传播下去最终写进 Kafka。举个我实际遇到的例子一个时间格式化 UDF输入是TIMESTAMP输出是STRING。当输入为 null 时UDF 内部没做判断直接调用了DateTimeFormatter.format(null)抛异常被捕获后返回了 null。结果就是所有时间字段为 null 的记录格式化后都是 null。这类问题的排查方法是在 UDF 里加日志把输入和输出都打出来看看 null 是从哪一步开始出现的。5.2 JOIN 操作产生的 null如果链路里有 JOIN那 null 的来源就更多了。LEFT JOIN 在右表没匹配到的时候右表字段全是 null这是正常语义。但如果你用的是 INNER JOIN理论上不该出 null除非 JOIN 条件本身涉及 null 字段导致匹配行为异常。还有一种情况是时态表 JOINTemporal Join如果版本表里没有对应版本的数据JOIN 出来的字段也会是 null。这种 null 是业务语义上的暂无数据不是 bug但下游如果不区分就会误以为是数据问题。5.3 类型转换失败的静默 nullFlink SQL 在做类型转换时如果转换失败默认行为是返回 null而不是报错。比如CAST(abc AS BIGINT)结果是 null。这个特性很坑因为它不会给你任何提示数据就悄悄变成 null 了。要避免这个问题可以在 SQL 里用TRY_CAST显式处理或者在作业配置里开启严格的类型检查。我的习惯是所有涉及类型转换的地方都加一层校验宁可让作业报错也别让脏数据流下去。6. 根因四Kafka Sink 的配置陷阱6.1 topic 和分区配置Kafka Sink 的配置里有几个参数会间接导致 null 问题。比如sink.partitioner如果配置成按某个字段分区而这个字段是 null那消息可能会被丢到默认分区或者直接失败。还有transaction.timeout.ms如果事务超时消息可能被回滚下游看到的就是空。这些配置本身不会直接产生 null但会让 null 问题变得更难排查。所以我的建议是排查阶段先把这些高级配置都关掉用最简配置跑一遍确认基础链路没问题再逐步加配置。6.2 exactly-once 语义下的可见性问题如果你开了 exactly-onceKafka Sink 会用事务写消息。事务未提交之前下游消费者是看不到数据的。如果事务因为某种原因一直没提交下游就会觉得数据没来而不是数据是 null。这两种现象要区分开。排查方法是看 Kafka 的事务状态或者临时把语义降级成 at-least-once看看数据是否正常。如果降级后正常那问题就在事务配置上不在数据本身。6.3 消息体过大导致的截断这个比较少见但确实遇到过。如果单条消息体超过了 Kafka 的max.request.size或者 broker 的message.max.bytes消息会被拒绝或者截断。截断后的 JSON 解析失败下游可能把整个消息当成 null 处理。这种情况的特征是只有大字段比如长文本出问题小字段正常。如果你观察到这个规律就去检查消息体大小限制。7. 完整排查流程与实操步骤7.1 第一步确认 null 的类型和范围先别急着改代码把 Kafka 里的原始消息捞出来看。用命令行消费几条kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic your_topic \ --from-beginning \ --max-messages 10看消息体里 null 是字段级、整条级还是缺失级。同时统计一下 null 的比例是全量 null 还是部分 null。全量 null 通常是配置问题部分 null 通常是数据问题。7.2 第二步在 Sink 之前加观测点把 Kafka Sink 临时换成 print sink或者加一个侧输出流把数据打到日志里。对比 print 出来的数据和 Kafka 里的数据如果 print 正常而 Kafka 不正常问题就在 Sink如果 print 就不正常问题在上游。这一步是分水岭能帮你把排查范围缩小一半。7.3 第三步逐字段核对 Schema把源表 Schema、中间转换的 Schema、Sink 表 Schema 三份列出来逐字段核对名字、类型、顺序。这一步最枯燥但最有效。我一般会用表格整理源字段源类型中间字段中间类型Sink字段Sink类型是否一致idBIGINTidBIGINTidBIGINT是user_nameVARCHARnameSTRINGnameSTRING是amountDECIMALamountDECIMALamountDECIMAL是只要有一行不一致就重点查那一行。7.4 第四步检查序列化器配置确认 Sink 的format配置是json、debezium-json还是自定义。如果是json检查有没有配置json.encode.decimal-as-plain-number之类的参数。如果是自定义序列化器把代码拉出来逐行看重点看 null 处理和类型转换。7.5 第五步验证修复效果改完之后别急着上生产。先在测试环境跑一批数据对比修复前后的 Kafka 消息。确认 null 问题消失且其他字段没受影响。然后再灰度上生产观察一段时间。8. 常见问题速查表与避坑经验8.1 常见问题速查表现象可能原因排查方法解决方案全量字段 nullSchema 顺序错位核对字段顺序显式写列名部分字段 nullUDF 返回 null在 UDF 加日志UDF 内做 null 判断字段缺失序列化器过滤 null检查序列化配置配置保留 null 字段整条 null反序列化失败看异常日志修复格式配置运行一段时间后出 nullSchema 演进检查表结构变更开启 Schema 演进大字段 null消息体超限检查消息大小调大限制或拆分8.2 避坑经验一永远显式写列名这条我重复三遍都不为过。Flink SQL 的位置匹配机制是 null 问题的头号元凶尤其是字段多、有计算字段的场景。显式写列名虽然啰嗦但能避免 90% 的错位问题。8.3 避坑经验二UDF 必须做 null 防御写 UDF 的时候第一行就该是 null 判断。输入为 null 时要么返回 null明确语义要么返回默认值绝对不能让它抛异常或者产生意外行为。我现在的习惯是每个 UDF 开头都写if (input null) { return null; }8.4 避坑经验三测试环境要覆盖 null 场景很多人测试的时候只用正常数据结果一上生产就出问题。测试数据里必须包含 null 字段、空字符串、超长文本、特殊字符把这些边界情况都跑一遍才能保证链路的健壮性。8.5 避坑经验四监控 null 比例生产环境上建议加一个监控指标统计 Kafka 消息里 null 字段的比例。一旦比例异常升高立刻告警。这样能在问题扩大之前发现它。我一般用 Flink 的 Metrics 系统自定义一个 Counter在序列化器里统计 null 数量。9. 我个人的几点实操体会排查 null 问题这件事最忌讳的就是猜。我见过太多人一上来就改序列化器改了半天发现是 SQL 写错了。正确的姿势是先观测、再定位、后修复每一步都要有数据支撑。另外Flink CDC Kafka 这条链路配置项非常多任何一个配置不当都可能引发 null 问题。我的建议是把链路拆成Source、Transform、Sink三段每段单独验证确认没问题再串起来。这样出问题的时候能快速定位到是哪一段。最后分享一个小技巧如果你用的是 Flink SQL可以在 Sink 之前加一个SELECT * FROM ... WHERE id IS NULL OR name IS NULL的过滤把有 null 的记录单独打到一个侧表里。这样既能保证主链路数据干净又能保留问题数据用于排查。这个做法我在好几个项目里都用过效果很好。还有一个容易被忽略的点是版本兼容性。Flink CDC 连接器、Flink 本体、Kafka 客户端这三者的版本如果搭配不当也可能出现序列化异常导致 null。我建议在项目初期就把版本矩阵固定下来写进文档避免后期升级时踩坑。升级任何一个组件之前都要在测试环境完整跑一遍链路验证。这套排查思路不只适用于 null 问题其他数据不一致的问题也能套用。核心就一句话让数据在每一层都可见问题自然无处遁形。