1. 问题背景与整体排查思路Flink CDC 这套链路在实时数仓里用得非常多典型场景就是 MySQL 业务库的 binlog 通过 Flink CDC 捕获后写入 Kafka 作为 ODS 层下游再对接实时计算或数据湖。链路本身不复杂但真正跑起来之后null 值问题几乎是最常见、也最让人头疼的一类故障。我最近就处理了一个案例某业务表通过 Flink CDC 同步到 Kafka下游消费时发现部分字段全是 null但源库里这些字段明明有值。这个问题的迷惑性在于它不像任务直接报错那样干脆。任务能正常跑Checkpoint 也能成功Kafka 里也有数据但就是某些字段的值丢了。排查起来需要从源端、Flink 作业配置、序列化格式、Kafka 消息结构四个层面逐层剥离。我处理下来最大的体会是null 问题很少是单一原因造成的往往是几个配置细节叠加的结果。先说一下这个案例的基本情况。源表是一张订单表大概 20 多个字段包含 varchar、int、decimal、datetime、tinyint 等多种类型。Flink 版本是 1.17CDC 连接器用的是 2.4 版本Kafka 连接器是 3.0 版本。作业用 Flink SQL 编写通过 CTAS 或 INSERT INTO 的方式把 CDC 源表数据写入 Kafka Sink。下游用 Flink SQL 消费 Kafka 时发现order_amount、create_time、status这几个字段全是 null。排查这类问题我的习惯是先定位 null 出现在哪个环节而不是一上来就改配置。具体来说分三步走第一步确认源库数据本身是否有值排除源端问题第二步在 Flink 作业里加一个临时 Print Sink 或者用 Flink SQL Client 直接查 CDC 源表看 Flink 读到的数据是否完整第三步检查 Kafka 里的原始消息看是写入时就是 null还是下游解析时变成 null。这三步能把问题范围从整条链路缩小到某一个具体环节。我见过很多人一上来就怀疑 CDC 连接器结果折腾半天发现是下游反序列化时 schema 对不上。所以定位环节这一步绝对不能省。提示排查 null 问题时一定要保留 Kafka 原始消息。用kafka-console-consumer直接消费看 JSON 原文比在下游 Flink 作业里 debug 高效得多。2. 核心原因拆解与逐项验证2.1 源端 DDL 与 Flink 表结构不一致导致的 null这是最常见的原因之一。Flink CDC 在启动时会根据源表的 schema 生成对应的 Flink 表结构但如果源表在作业启动后发生了 DDL 变更比如新增字段、修改字段类型而 Flink 作业没有重启或没有开启 schema 演进就会出现字段错位或缺失。我这个案例里源表在作业运行期间新增了一个coupon_id字段但 Flink 作业的 schema 还是旧的。结果就是新字段在 Flink 侧不存在写入 Kafka 时该字段为 null。更隐蔽的情况是字段顺序变化如果源表做了ALTER TABLE ... MODIFY COLUMN调整了列顺序而 Flink 侧没有同步就会出现张冠李戴——A 字段的值写到了 B 字段的位置看起来像是某些字段为 null。验证方法很简单在 Flink SQL Client 里执行DESCRIBE source_table和源库的SHOW CREATE TABLE对比逐字段核对名称、类型、顺序。如果发现不一致要么重启作业让 CDC 重新抓取 schema要么在 Flink 表定义里显式声明 schema 并开启scan.incremental.schema.enrichment.enabled。2.2 序列化格式与 Kafka 消息体不匹配Flink 写入 Kafka 时format的选择直接决定了消息体的结构。常见的有json、debezium-json、canal-json、avro等。如果 Sink 端用的是json格式但 Flink 表里某些字段没有正确映射或者下游用debezium-json去解析json格式的消息就会出现字段解析为 null。我遇到过一种典型情况Sink 端配置了debezium-json但源表是 MySQL CDC消息体里包含before、after、op等字段。下游消费时如果直接按普通 JSON 解析取after里的字段但代码里写的是before那自然全是 null。还有一种情况是json格式下Flink 默认会把 null 字段也序列化进去但如果下游 schema 里字段名大小写不一致也会解析失败。验证方法用kafka-console-consumer消费几条消息看原始 JSON 结构。重点看三个地方——字段名是否和下游 schema 一致、嵌套结构是否正确、null 字段是否真的在消息体里。2.3 主键与 Changelog 模式的影响Flink CDC 写入 Kafka 时如果源表有主键CDC 会生成 changelog 流包含I、-U、U、-D四种操作类型。如果 Sink 端没有正确处理这些操作类型或者下游消费时没有按 changelog 语义解析就会出现数据丢失或 null。具体来说如果 Sink 端用的是upsert-kafka连接器它要求必须定义主键并且只支持INSERT和UPDATE语义。如果源表没有主键或者主键定义不对写入时就会出问题。另外upsert-kafka在序列化时key 和 value 是分开的如果 key 为 nullvalue 也可能写入失败或变成 null。我这个案例里源表是有主键的但 Sink 端用的是普通kafka连接器而不是upsert-kafka导致 changelog 流里的-U和-D消息也被写入了 Kafka。下游消费时如果没有过滤这些操作类型就会看到大量 null 或重复数据。2.4 数据类型映射与精度丢失Flink 和 MySQL 的数据类型映射不是一一对应的某些类型在转换过程中会丢失精度或变成 null。比如 MySQL 的decimal(20,10)映射到 Flink 的DECIMAL(20,10)如果 Flink 表里定义成了DECIMAL(10,2)超出精度的值就会变成 null。再比如datetime类型如果 Flink 表里定义成了TIMESTAMP(3)而源库有微秒精度也可能出问题。还有一种情况是tinyint(1)在 MySQL 里常用来表示布尔值但 Flink 里如果映射成了TINYINT而不是BOOLEAN下游解析时可能因为类型不匹配变成 null。这个坑我在多个项目里都踩过尤其是下游用 Java 或 Scala 消费时类型转换异常会被吞掉最终表现为 null。验证方法在 Flink SQL Client 里执行SELECT * FROM source_table LIMIT 10看每个字段的值是否正常。如果 Flink 侧读出来就是 null那问题在 CDC 或类型映射如果 Flink 侧正常但 Kafka 里是 null那问题在 Sink 序列化。2.5 Kafka Sink 的 null 值处理策略Flink Kafka Sink 对 null 值的处理有几个关键配置。json格式下默认会忽略 null 字段但如果配置了json.ignore-parse-errors为false遇到 null 可能会直接报错。另外debezium-json格式下before和after字段的结构是固定的如果源表某些字段为 null消息体里会显式写成null下游解析时需要注意。还有一个容易被忽略的点Kafka 消息的 key。如果 Sink 端配置了key.format但源表主键为 null比如联合主键里某个字段为 null写入时 key 就是 null某些下游消费者会把 key 为 null 的消息直接丢弃或标记为异常。3. 完整排查流程与实操步骤3.1 环境准备与复现步骤要复现和排查这个问题我建议搭一个最小化的本地环境。需要准备的东西不多一个 MySQL 实例可以用 Docker 起、一个 Kafka 实例单节点即可、一个 Flink 集群本地 Standalone 或 MiniCluster 都行。版本方面Flink 1.17 CDC 2.4 Kafka 3.0 是比较稳定的组合。具体步骤如下在 MySQL 里建一张测试表包含多种数据类型并插入几条测试数据在 Flink SQL Client 里创建 CDC 源表和 Kafka Sink 表执行INSERT INTO sink SELECT * FROM source用kafka-console-consumer消费 Kafka 消息观察字段值对比源库数据、Flink 查询结果、Kafka 消息体定位 null 出现的环节。这个流程看起来简单但每一步都有细节。比如创建 CDC 源表时connector要选mysql-cdcscan.startup.mode建议用initial先全量再增量方便对比。Kafka Sink 表的format先用json方便肉眼观察。3.2 Flink SQL 表定义的关键参数CDC 源表的定义里有几个参数直接影响 null 问题CREATE TABLE mysql_source ( id BIGINT, order_no VARCHAR(64), order_amount DECIMAL(20,10), create_time TIMESTAMP(3), status TINYINT, PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname localhost, port 3306, username root, password 123456, database-name test_db, table-name orders, scan.incremental.snapshot.enabled true, scan.incremental.schema.enrichment.enabled true, debezium.snapshot.mode initial );这里scan.incremental.schema.enrichment.enabled设为true很关键它能让 Flink 在运行期间感知源表的 schema 变更。debezium.snapshot.mode设为initial表示先做全量快照再读增量避免启动时漏数据。Kafka Sink 表的定义CREATE TABLE kafka_sink ( id BIGINT, order_no VARCHAR(64), order_amount DECIMAL(20,10), create_time TIMESTAMP(3), status TINYINT, PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector upsert-kafka, topic orders_topic, properties.bootstrap.servers localhost:9092, key.format json, value.format json, value.json.fail-on-missing-field false, value.fields-include ALL );注意这里用的是upsert-kafka而不是普通kafka因为源表有主键用upsert-kafka能正确处理 changelog。value.json.fail-on-missing-field设为false可以避免字段缺失时直接报错。value.fields-include设为ALL表示 key 和 value 里都包含所有字段方便下游解析。3.3 参数计算与选择依据DECIMAL(20,10)这个精度不是随便定的。源库的order_amount字段是decimal(20,10)所以 Flink 侧必须保持一致。如果 Flink 侧定义成DECIMAL(10,2)那么超过 10 位整数或 2 位小数的值就会溢出表现为 null 或截断。计算方式很简单整数位 小数位 总精度。源库是 20 位总精度、10 位小数Flink 侧就必须是DECIMAL(20,10)。TIMESTAMP(3)的精度也要注意。MySQL 的datetime默认精度是 0秒级但如果是datetime(6)就是微秒级。Flink 的TIMESTAMP(3)是毫秒级TIMESTAMP(6)是微秒级。如果源库是微秒级而 Flink 侧用了TIMESTAMP(3)微秒部分会被截断但不会变成 null。真正会变成 null 的情况是源库有0000-00-00 00:00:00这种非法日期Flink 解析时会报错或置为 null。TINYINT和BOOLEAN的映射也要注意。MySQL 的tinyint(1)在 Flink CDC 里默认映射成TINYINT但如果下游期望的是BOOLEAN就需要在 Flink 表里显式定义成BOOLEAN或者在下游做类型转换。我一般建议在 Flink 侧就定义成BOOLEAN避免下游再处理。3.4 Kafka 消息体结构验证写入 Kafka 后用下面的命令消费消息kafka-console-consumer --bootstrap-server localhost:9092 \ --topic orders_topic \ --from-beginning \ --property print.keytrue \ --property key.separator | 正常情况下的消息体应该是这样的{id:1,order_no:ORD20240101001,order_amount:99.99,create_time:2024-01-01T10:00:00,status:1}如果看到某个字段是null就要对比源库数据。比如order_amount在源库是99.99但 Kafka 里是null那问题就在 Flink 的序列化环节。如果 Kafka 里字段名是orderAmount而不是order_amount那就是命名策略问题下游按order_amount取就会得到 null。还有一种情况是消息体里嵌套了before和after{before:null,after:{id:1,order_no:ORD20240101001,order_amount:99.99},op:c}这是debezium-json格式的典型结构。下游如果直接取顶层字段就会全是 null必须取after里的字段。4. 常见问题速查与避坑经验4.1 null 问题速查表现象可能原因排查方法解决方案所有字段全为 null下游解析格式与 Sink 格式不一致查看 Kafka 原始消息结构统一 format或下游按实际结构调整部分字段为 null源表 DDL 变更未同步对比 Flink 表结构与源表 DDL重启作业或开启 schema 演进数值字段为 null精度定义不匹配对比源库与 Flink 的 DECIMAL 精度保持精度一致时间字段为 null非法日期或精度不匹配检查源库是否有 0000-00-00清洗数据或调整 TIMESTAMP 精度主键字段为 null联合主键部分字段为空检查源表主键定义确保主键字段非空changelog 消息导致 null未使用 upsert-kafka查看消息体是否有 op 字段改用 upsert-kafka 或下游过滤4.2 实操避坑心得第一个坑是不要用普通 kafka 连接器接 CDC 源表。CDC 源表是 changelog 流包含-U和-D操作。普通kafka连接器会把这些操作也写进去下游如果不处理就会看到大量 null 或重复数据。用upsert-kafka连接器它会自动根据主键做 upsert只保留最新状态。第二个坑是schema 演进不是万能的。虽然scan.incremental.schema.enrichment.enabled能感知 DDL 变更但它只支持新增字段不支持删除字段或修改字段类型。如果源表删了字段Flink 侧还是会保留写入 Kafka 时该字段为 null。所以生产环境里源表 DDL 变更一定要和 Flink 作业联动。第三个坑是Kafka 消息的 key 不要忽略。用upsert-kafka时key 是主键的序列化结果。如果主键字段为 nullkey 就是 null某些下游消费者会直接丢弃。所以源表主键一定要非空联合主键的每个字段都要非空。第四个坑是下游消费时不要用SELECT *。Kafka 消息体里的字段顺序和 Flink 表定义可能不一致用SELECT *容易错位。建议显式列出字段名并且和 Flink 表定义保持一致。4.3 生产环境建议生产环境里我建议在 Flink 作业和 Kafka 之间加一层校验。具体做法是在 Sink 之前加一个临时 Print Sink 或者用 Flink Metrics 监控 null 字段的数量。如果某个字段的 null 率突然升高就能及时发现。另外Kafka 的 topic 建议按业务分不要所有表都写一个 topic。不同表的 schema 不一样混在一起容易出问题。如果确实要合并建议用debezium-json格式它自带 schema 信息下游解析时更安全。还有一点是版本兼容性。Flink CDC 和 Kafka 连接器的版本要匹配CDC 2.4 建议配 Flink 1.17Kafka 连接器 3.0。版本不匹配时序列化行为可能有差异null 问题会更难排查。我一般会在测试环境先跑一遍全链路确认没问题再上生产。5. 问题复现与修复验证5.1 最小化复现案例为了验证上面的分析我搭了一个最小化复现环境。源表定义如下CREATE TABLE orders ( id BIGINT PRIMARY KEY, order_no VARCHAR(64), order_amount DECIMAL(20,10), create_time DATETIME(3), status TINYINT(1) );插入一条测试数据INSERT INTO orders VALUES (1, ORD001, 99.99, 2024-01-01 10:00:00.123, 1);Flink Sink 表先用普通kafka连接器format用json。执行写入后消费 Kafka 消息发现order_amount和create_time是 null。原因就是普通kafka连接器不处理 changelog-U和-D消息里的字段为 null。改成upsert-kafka连接器后重新写入消息体变成{id:1,order_no:ORD001,order_amount:99.99,create_time:2024-01-01T10:00:00.123,status:1}所有字段都正常了。5.2 修复方案对比方案适用场景优点缺点改用 upsert-kafka源表有主键自动处理 changelog消息干净要求主键非空下游过滤 changelog源表无主键灵活下游逻辑复杂容易漏开启 schema 演进源表频繁 DDL自动同步 schema只支持新增字段显式声明 schema源表结构稳定可控性强DDL 变更需手动同步5.3 验证清单修复后我一般会按下面的清单逐项验证源库数据是否完整有无 null 值Flink SQL Client 查询 CDC 源表字段值是否正常Kafka 原始消息体字段名和值是否正常下游消费作业解析后的字段是否正常监控 null 率确认没有异常升高。这个清单看起来繁琐但能覆盖整条链路避免修了一个环节又漏了另一个。我踩过的坑里有一半是因为只验证了 Flink 侧没验证 Kafka 侧结果下游还是 null。6. 个人实操体会与后续扩展处理完这个案例我最大的体会是null 问题本质上是 schema 一致性问题。从源库到 Flink 到 Kafka 再到下游每一层的 schema 都必须严格对齐。任何一层的字段名、类型、顺序、精度有差异都可能表现为 null。所以排查时不要只盯着一个环节要整条链路一起看。另外Flink CDC 的 schema 演进功能虽然好用但不能完全依赖。生产环境里源表 DDL 变更一定要有流程管控变更前通知下游变更后验证数据。我见过太多因为 DDL 变更导致下游数据异常的案例最后都是靠人工补数据解决的。后续如果要做扩展我建议在 Flink 作业里加一个数据质量校验模块。比如用 Flink SQL 的COUNT和SUM统计每个字段的 null 率写入监控系统。这样一旦 null 率异常就能第一时间告警而不是等下游反馈。这个模块实现起来不难但能省很多排查时间。最后分享一个小技巧如果怀疑是序列化问题可以先把 Sink 的format改成debezium-json它自带 schema 信息消息体里会显式标注每个字段的类型。虽然消息体会大一些但排查问题时非常直观。确认没问题后再改回json兼顾性能和可读性。