Flink CDC MongoDB Connector 的实现原理和使用实践

摘要：本文整理自 XTransfer 资深 Java 开发工程师、Flink CDC Maintainer 孙家宝在 Flink CDC Meetup 的演讲。主要内容包括：

MongoDB Change Stream 技术简介MongoDB CDC Connector 业务实践MongoDB CDC Connector 生产调优MongoDB CDC Connector 并行化 Snapshot 改进后续规划

01MongoDB Change Stream 技术简介

MongoDB 是一种面向文档的非关系型数据库，支持半结构化数据存储；也是一种分布式的数据库，提供副本集和分片集两种集群部署模式，具有高可用和水平扩展的能力，比较适合大规模的数据存储。另外， MongoDB 4.0 版本还提供了多文档事务的支持，对于一些比较复杂的业务场景更加友好。

MongoDB 使用了弱结构化的存储模式，支持灵活的数据结构和丰富的数据类型，适合 Json 文档、标签、快照、地理位置、内容存储等业务场景。它天然的分布式架构提供了开箱即用的分片机制和自动 rebalance 能力，适合大规模数据存储。另外， MongoDB 还提供了分布式网格文件存储的功能，即 GridFS，适合图片、音频、视频等大文件存储。

MongoDB 提供了副本集和分片集两种集群模部署模式。

副本集：高可用的部署模式，次要节点通过拷贝主要节点的操作日志来进行数据的复制。当主要节点发生故障时，次要节点和仲裁节点会重新发起投票来选出新的主要节点，实现故障转移。另外，次要节点还能分担查询请求，减轻主要节点的查询压力。

分片集：水平扩展的部署模式，将数据均匀分散在不同 Shard 上，每个 Shard 可以部署为一个副本集，Shard 中主要节点承载读写请求，次要节点会复制主要节点的操作日志，能够根据指定的分片索引和分片策略将数据切分成多个 16MB 的数据块，并将这些数据块交给不同 Shard 进行存储。Config Servers 中会记录 Shard 和数据块的对应关系。

MongoDB 的 Oplog 与 MySQL 的 Binlog 类似，记录了数据在 MongoDB 中所有的操作日志。Oplog 是一个有容量的集合，如果超出预设的容量范围，则会丢弃先前的信息。

与 MySQL 的 Binlog 不同， Oplog 并不会记录变更前/后的完整信息。遍历 Oplog 的确可以捕获 MongoDB 的数据变更，但是想要转换成 Flink 支持的 Changelog 依然存在一些限制。

首先，订阅 Oplog 难度较大。每个副本集会维护自己的 Oplog， 对于分片集群来说，每个 Shard 可能是一个独立的副本集，需要遍历每个 Shard 的 Oplog 并按照操作时间进行排序。另外， Oplog 没有包含变更文档前和变更后的完整状态，因此既不能转换成 Flink 标准的 Changelog ，也不能转换成 Upsert 类型的 Changelog 。这亦是我们在实现 MongoDB CDC Connector 的时候没有采用直接订阅 Oplog 方案的主要原因。

最终我们选择使用 MongoDB Change Streams 方案来实现 MongoDB CDC Connector。

Change Streams 是 MongoDB 3.6 版本提供的新特性，它提供了更简单的变更数据捕获接口，屏蔽了直接遍历 Oplog 的复杂度。Change Streams 还提供了变更后文档完整状态的提取功能，可以轻松转换成 Flink Upsert 类型的 Changelog。它还提供了比较完整的故障恢复能力，每一条变更记录数据都会包含一个 resume token 来记录当前变更流的位置。故障发生后，可以通过 resume token 从当前消费点进行恢复。

另外， Change Streams 支持变更事件的筛选和定制化的功能。比如可以将数据库和集合名称的正则过滤器下推到 MongoDB 来完成，可以明显减少网络开销。它还提供了对集合库以及整个集群级别的变更订阅，能够支持相应的权限控制。

使用 MongoDB Change Streams 特性实现的 CDC Connector 如上图所示。首先通过 Change Streams 订阅 MongoDB 的变更。比如有 insert、update、delete、replace 四种变更类型，先将其转换成 Flink 支持的 upsert