Flink CDC 在大健云仓的实践

公司引入 CDC 技术，主要基于以下四个角色的需求：

• 物流科学家：需要库存、销售订单、物流账单等数据用于做分析。
• 开发：需要同步其他业务系统的基本信息。
• 财务：希望财务数据能够实时传送到财务系统，而不是月结前才能看到。
• 老板：需要数据大屏，通过大屏查看公司的业务和运营情况。

• 基于查询：查询后插入、更新到数据库即可，无须数据库的特殊配置以及账号权限。它的实时性基于查询频率决定，只能通过提高查询频率来保证实时性，而这必然会对 DB 造成巨大压力。此外，因为是基于查询，所以它无法捕获两次查询之间数据的变更记录，也就无法保证数据的一致性。
• 基于日志：通过实时消费数据的变更日志实现，因此实时性很高。而且不会对 DB 造成很大的影响，也能够保证数据的一致性，因为数据库会将所有数据的变动记录在变更日志中。通过对日志的消费，即可明确知道数据的变化过程。它的缺点是实现相对复杂，因为不同数据库的变动日志实现不一样，格式、开启方式以及特殊权限都不一样，需要针对每一种数据库做相应的适配开发。

• 数据源：Flink CDC 除了对传统的关系型数据库做到了很好的支持外，对文档型、NewSQL（TiDB、OceanBase） 等当下流行的数据库都能够支持；Debezium 对数据库的支持相对没有那么广泛，但是对主流的关系型数据库都做到了很好的支撑；Canal 和 OGG 只支持单一的数据源。
• 断点续传：四种技术都能够支持。
• 同步模式：除了 Canal 只支持增量，其他技术均支持全量 + 增量的方式。而全量 + 增量的方式意味着第一次上线时全量到增量的切换过程全部可以通过 CDC 技术实现，无须人为地通过全量的任务加上增量的 job 去实现全量 + 增量数据的读取。
• 活跃度：Flink CDC 拥有非常活跃的社区，资料丰富，官方也提供了详尽的教程以及快速上手教程；Debezium 社区也相当活跃，但资料大多是英文的；Canal 的用户基数特别大，资料也相对较多，但社区活跃度一般；OGG 是 Oracle 的大数据套件，需要付费，只有官方资料。
• 开发难度：Flink CDC 依靠 Flink SQL 和 Flink DataStream 两种开发模式，尤其是 Flink SQL，通过非常简单的 SQL 即可完成数据同步任务的开发，开发上手尤为简单；Debezium 需要自己解析采集到的数据变更日志进行单独处理，Canal 亦是如此。
• 运行环境依赖：Flink CDC 是以 Flink 作为引擎，Debezium通常是将 Kafka connector 作为运行容器；而 Canal 和 OGG 都是单独运行。
• 下游丰富程度：Flink CDC 依靠 Flink 非常活跃的周边以及丰富的生态，能够打通丰富的下游，对普通的关系型数据库以及大数据存储引擎 Iceberg、ClickHouse、Hudi 等都做了很好的支持；Debezium 有 Kafka JDBC connector， 支持 MySQL 、Oracle 、SqlServer；Canal 只能直接消费数据或将其输出到 MQ 中进行下游的消费；OGG 因为是官方套件，下游丰富程度不佳。

• 2018 年之前，大健云仓数据同步的方式为：通过多数据应用定时同步系统之间的数据。
• 2020 年之后，随着跨境业务的飞速发展，多数据源应用经常打满 DB 影响在线应用，同时定时任务的执行顺序管理混乱。
• 因此， 2021 年我们开始调研选型 CDC 技术，搭建了小型试验场景，进行小规模的试验。
• 2022 年，上线了基于 Flink CDC 实现的 LDSS 系统库存场景同步功能。
• 未来，我们希望依托 Flink CDC 打造数据同步平台，通过界面的开发和配置完成同步任务的开发、测试和上线，能够全程在线管理同步任务的整个生命周期。

• 仓储部门：要求仓库的库存容量和商品品类分布合理，库存容量方面，需要留一些 buffer 以防突如其来的入库单导致爆仓；商品品类方面，季节性的商品库存分配不合理导致热点问题，这必将给仓库的管理带来巨大挑战。
• 平台客户：希望订单处理及时，货物能够快速、精准地交到客户手上。
• 物流部门：希望能够提升物流效率，降低物流成本，高效利用有限的运力。
• 决策部门：希望 LDSS 系统能够对在何时何地新建仓库提供科学的建议。

1. 第一步，定义源表 —— 需要同步的表；
2. 第二步，定义目标表 —— 需要写入数据的目标表；
3. 第三步，通过 insert select 语句，即可完成 CDC 同步任务的开发。

1. 能够很好地管理数据源、表等元信息；
2. 任务的整个生命周期都可以在平台上完成；
3. 实现任务的性能观测以及告警；
4. 简化开发，快速上手，业务开发人员经过简单培训即可上手开发同步任务。

1. 收拢数据同步任务，统一来管理；
2. 平台管理维护同步任务的全生命周期；
3. 专门的团队负责，团队能够专注前沿的数据集成技术。

• 实时数仓：希望通过 Flink CDC 以支持更多实时数仓的业务场景，借助 Flink 强大的计算能力做一些数据库的物化视图。将计算从 DB 里解脱出来，通过 Flink 的外部计算再重新写回数据库，以加速平台应用的报表、统计、分析等实时应用场景。
• 实时应用：Flink CDC 能够从 DB 层捕获变更，因此可以通过 Flink CDC 实时更新搜索引擎中的内容，实时向财务系统推送财务和核算数据。因为大部分财务系统的数据都需要业务系统通过跑定时任务以及经过大量关联、聚合、分组等操作才能计算出来，再推送到财务系统中。而借助 Flink CDC 强大的数据捕获能力，再加上 Flink 的计算能力，将这些数据实时地推送到核算系统和财务系统，就能够及时发现业务的问题，减少公司的损失。
• 缓存：通过 Flink CDC，能够构建一个脱离于传统的应用之外的实时缓存，对于在线应用的性能有极大的提升。

• 第一，开源共建。希望能够有更多机会与同行交流分享 Flink CDC 在公司落地实践的经验以及接入的场景，也会在内部开展培训 Flink CDC 技术，通过培训让大家了解 Flink CDC 技术，并在实际工作中能够通过这项技术来解决更多的业务痛点。
  目前公司和社区的合作共建已取得一些成果，公司向社区贡献了 SqlServer CDC Connector 以及合作完成了 TiDB CDC Connector。

• 第二，服务社区。培养部门开发的开源合作能力，并将公司内部版本的特性贡献给社区，只有经过社区广大用户的打磨，特性才能更加稳定合理。此外，也希望能够在 schema evolution、turning performance、整库同步的方向与社区开展紧密合作。
• 第三，探索方向。相信 Flink CDC 不会满足于当下的成就，必定会继续向更远的目标前进。所以希望能够与社区共同探索发掘 Flink CDC 更多可能的方向。

1. 快照过程中锁表：锁表操作对于 DBA 和在线应用都是不可忍受的， DBA 无法接受数据库被夯住，同时也会影响在线应用。
2. 快照过程中不能 checkpoint：不能 checkpoint 就意味着快照过程中一旦失败，只能重新开始跑快照过程，这对于大表非常不友好。
3. 快照过程只支持单并发：千万级、上亿级的大表，在单并发的情况下需要同步十几甚至几十个小时，极大束缚了 SqlServer CDC 的应用场景。

提问

Qustions

&

解答

Answers

Q1

需要开启 SqlServer 自己的 CDC 吗？

是的，SqlServer CDC 的功能就是基于 SqlServer 数据库自己的 CDC 特性实现的。

Q2

物化视图通过什么方式去刷新定时任务触发器？

通过 Flink CDC 将需要生成物化视图的 SQL 放在 Flink 里运行，通过原表的变动触发计算，然后同步到物化视图表里。

Q3

平台化是怎么做的？

平台化参考了社区众多的开源项目以及优秀的开源平台，比如 StreamX、DLink 等优秀的开源项目。

Q4

SqlServer CDC 在消费 transaction log 时有瓶颈吗？

SqlServer 并没有直接消费 log，其原理是 SqlServer capture process 去匹配 log 内哪些表开启了 CDC ，然后将这些表从日志里捞到开启 CDC 表的变更数据，再转插到 change table 里，最后通过开启 CDC 之后数据库生成的 CDC query function 获取到数据的变更。

Q5

Flink CDC 高可用如何保障同步任务过多或密集处理方案？

Flink 的高可用依赖于 Flink 特性比如 checkpoint 等来保证。同步任务过多或处理方案密集的情况，建议使用多套 Flink 下游集群，然后根据同步的实时性区分对待，将任务发布到相应的集群中。

Q6

中间需要 Kafka 吗？

取决于同步任务或数仓架构是否需要将中间数据做 Kafka 落地。

Q7

一个数据库中有多张表，可以放到一个任务里运行吗？

取决于开发方式。如果是 SQL 的开发方式，要实现一次性写多表只能通过多个任务。但 Flink CDC 提供了另外一种比较高阶的开发方式 DataStream ，可以将多表放到一个任务里运行。

Q8

Flink CDC 支持读取 Oracle 从库的日志吗？

目前还无法实现。

Q9

通过 CDC 同步后两个端的数据质量如何监控，如何比对？

目前只能通过定时抽样来做数据质量的检查，数据质量问题一直是业内比较棘手的问题。

Q10

大健云仓用的什么调度系统？系统如何与 Flink CDC 集合？

使用 XXL Job 作为分布式的任务调度，CDC 没有用到定时任务。

Q11

如果采集增删表，SqlServer CDC 需要重启吗？

SqlServer CDC 目前不支持动态加表的功能。

Q12

同步任务会影响系统性能吗？

基于 CDC 做同步任务肯定会影响系统性能，尤其是快照过程对数据库会有影响，进而影响应用系统。社区将来会做限流、对所有 connector 做并发无锁的实现，都是为了扩大 CDC 的应用场景以及易用性。

Q13

全量和增量的 savepoint 怎么处理？

（未通过并发无锁框架实现的连接器）全量过程中不可以触发 savepoint，增量过程中如果需要停机发布，可通过 savepoint 恢复任务。

Q14

CDC 同步数据到 Kafka ，而 Kafka 里面存的是 Binlog ，如何保存历史数据和实时数据？

将 CDC 同步的数据全部 Sync 到 Kafka，保留的数据取决于 Kafka log 的清理策略，可以全部保留。

Q15

CDC 会对 Binlog 的日志操作类型进行过滤吗？会影响效率吗？

即使有过滤操作，对性能影响也不大。

Q16

CDC 读 MySQL 初始化快照阶段，多个程序读不同的表会有程序报错无法获取锁表的权限，这是什么原因？

建议先查看 MySQL CDC 是不是使用老的方式实现，可以尝试新版本的并发无锁实现。

Q17

MySQL 上亿大表全量和增量如何衔接？

建议阅读雪尽老师在 2.0 的相关博客，非常简单清晰地介绍了并发无锁如何实现一致性快照，完成全量和增量的切换。

《Flink CDC 2.0 正式发布，详解核心改进》