数据增量同步检验问题

[aoxg2019]

如题，在 mysql 通过 canal 发送数据到 kafka 后，进行增量同步数据到 es ，现在的问题是如何检验增量同步过程中有没有问题呢？这个工作可能要周期运行！举例如下:此刻比对昨天以前的增量数据的总数，及抽样某个表的一些行来对比 es 中的数据。查 mysql 之前，数据无变化，查 es 期间，mysql 中某些数据已经被更新或者删除了，但数据更新还没写入 es ，这样查出来的数据比对结果不太一致。各位有什么好的方案吗？

[miniliuke]

@aoxg2019 不用看框架，直接看《 DBLog: A Watermark Based Change-Data-Capture Framework 》论文和解读就可以了

[lmshl]

我觉得业务层还是需要按照业务层的思路去解决，这和基础设施取舍不同。
比如按照做基础设施的思路来搞，那应该每一步都不可以出错，如果出错了就应该停在当前位置无限重试下去，以保证数据最终一致性。
但按照业务思路来做，因为一条数据出错而导致整个系统数据同步停机是不可接受的。

所以必然是以业务行（ row ）为单位，多次重试后记录错误并跳过，ES 也仅供搜索，业务事务依然由 RDBMS 保证，如此则需要引入就死信队列（ DLQ ）与纠错机制。

DLQ 能弥补一部分错误，但无法处理某些内部错误被当作正确处理跳过的场景。例如上游有 BUG ，请求账户积分失败时返回了 0 ，虽然锅是上游的，但修数据依然是下游要处理的。

所以还是需要有一种纠错机制来保证数据的最终一致性。我最近在考虑哈希树（ merkle tree ）不错，它是区块链用于校准的数据结构，可以快速对比不一致的数据块。


比如我们可以定时在闲暇对数据库做全量或部份 merkel tree 计算并对比两侧结果，最近数据多算，历史数据少算。这样对比出的不一致结果再通知给开发，找一下是哪里出的问题，以及手工对数据做补偿等等。

[aoxg2019]

@lmshl 感谢您提供的思路

[aoxg2019]

@miniliuke 好的，我看看这篇论文

[EXChen]

@EXChen 我们以前 binlog 往 ES 里面写数据是自己写代码实现的（其实是移植的 canal ），能够 cache 异常并告警处理。

[gengzi]

@miniliuke 大佬，想请教下，对于 Chunk 的选择，从主键 id 最小开始嘛？下一次选择上次主键最大后的下一个 Chunk ，直到所有的 chunk 选择完？关于全量数据校验，这个如何校验，查源表 Chunk 和 目标表 chunk 数据是否一致吗？

[potatowish]

Debezium 的增量临时快照可以看一下，基于《 DBLog: A Watermark Based Change-Data-Capture Framework 》论文的一个解决方案。

https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-incremental-snapshots
https://github.com/debezium/debezium-design-documents/blob/main/DDD-3.md