Hudi 生产问题排障-乱序Upsert入湖数据丢失

一、背景与问题在大数据流式处理领域乱序一直是一个无法越过的问题如何正确处理乱序数据也是流式组件不断努力优化的方向比如FLink提供的watermark机制forBoundedOutOfOrderness/allowedLateness/sideOutputLateData也是应对数据延迟乱序的设计。Hudi作为实时湖存储表格式提供流式数据的插入、更新、删除能力自身也对乱序有一定的处理策略特别是Upsert操作模式下支持基于precombine.field字段解决同主键数据冲突基于hoodie.is_deleted字段实现物理删除。最近我们遇到了一个头疼的问题有项目组反馈一张Flink1.16实时Upsert写入的Hudi0.14表数据量不对与离线Hive表数据做核验发现少了几万条。二、问题排查与分析我们首先检查FLink作业监控与运行日志发现指标监控与日志均正常没有任何反压、Exception等问题排除了作业运行异常导致的数据问题。我们找项目组调研了作业的数据链路与处理逻辑通过Canal实时采集上游Mysql的Binlog日志到KafkaFlink消费Kafka以Upsert模式写入Hudi表为了处理主键冲突配置了业务时间TMP_TIME为预合并字段。这是一个简单而常规的实时入湖链路理论上应该不会有问题。我们找了其他几张实时Upsert入湖的Hudi表与Hive表数据条数进行核验误差都在可控范围内实时与离线难以精准100%对齐不像出问题的这张表误差有几万条。接着推测会不会是乱序原因导致预合并失效找到Canal采集项目组确认Kafka-Topic推送策略对方反馈出于性能与负载均衡考虑采取的轮询推送策略同时发现该Kafka-Topic分区数为10。从技术上分析轮询多分区的写入搭配确实容易出现同一条主键的数据发往多个分区出现乱序处理场景。基于此思路我们在测试环境做了乱序模拟验证模拟更新删除乱序下预合并的表现。数据顺序到达或仅插入场景处理结果符合预期正常。数据乱序到达仅更新场景因建表未设置乱序处理策略默认按照OverwriteWithLatestAvroPayload即按最新到达覆盖旧数据设置为EventTimeAvroPayload可按预合并字段值排序处理。数据乱序到达涉及更新删除场景建表设置乱序处理策略为EventTimeAvroPayloadString与Decimal类型预合并字段因类型不匹配导致处理排序不生效。我们将问题表的预合并字段类型修改为bigint或timestamp进行验证后Hudi预合并处理符合预期。接着我们进行端到端数据验证发现在更新删除场景下仍然存在数据对不上的问题除了预合并阶段我们思考是否还有其他环节也存在处理异常呢我们查阅官方文档以及源码Hudi在处理同主键数据的变更合并时涉及三个关键阶段每个阶段都可能成为问题的源头预合并阶段当多条相同主键的记录同时出现在内存缓冲区时Hudi会在写入前调用HoodieRecordPayload#preCombine进行去重合并。该过程仅保留预合并字段值最大的记录此阶段对应调用链EventTimeAvroPayload.preCombine→ compareTo。读时合并阶段在MOR表读取时当Log文件中存在删除记录而Base Parquet文件中存在对应插入记录时Hudi需要通过HoodieMergedLogRecordScanner#processNextDeletedRecord来协调二者该过程需依据预合并字段值判断是保留插入记录还是应用删除记录。合并阶段在Compaction表合并阶段当Log和Parquet文件中存在同主键记录时通过EventTimeAvroPayload#combineAndGetUpdateValue进行最终合并。我们查看Compaction源码发现在处理delete记录合并时Hudi以删除记录优先未进行预合并字段比较排序。到这里问题表的异常原因基本就清晰了是Hudi三个合并处理阶段对乱序删除场景下存在逻辑处理异常。上述问题根因汇总如下序号问题阶段根本原因影响范围1预合并String/Decimal类型比较时类不匹配Utf8 vs StringGenericData.Fixed vs BigDecimal预合并结果不符合预期2读时合并processNextDeletedRecord中String/Decimal类型比较时类不匹配删除记录与插入记录取舍错误3合并combineAndGetUpdateValue中删除优先原则未考虑预合并字段值比较删除操作可能误删更新记录三、解决方案最直接的解决方案是修复Hudi在三个合并阶段的源码逻辑BUG能正常处理相同主键的记录判断与取舍操作。此问题已记录Issue#17642修复pr#17713影响0.14/1.x等多个版本暂未合入Release。此问题是相同主键乱序处理引发的保障数据顺序到达Hudi可以规避。Canal采集策略由轮询改造为Hash保障局部有序。相同主键数据有序在Flink写入Hudi前增加排序预处理。如row_num等最终我们通过修改源码重新编译打包上线解决灰度验证准出通过另一方面也在推动整体的Canal采集策略调整经过理论分析与数据验证具备可行性。四、总结展望本次生产问题的排查过程是一次由表及里的诊断剖析对账-核验-溯源。Hudi的Payload机制和preCombine机制虽然功能强大但深入理解其在不同场景下的行为差异至关重要。在后续使用过程中我们也要不断完善测试用例、监控体系及时跟进社区技术发展新动态推动Hudi技术栈的稳定高效使用。