关于flink:Flink-CDC-MongoDB-Connector-的实现原理和使用实践

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

1. MongoDB Change Stream 技术简介
2. MongoDB CDC Connector 业务实际
3. MongoDB CDC Connector 生产调优
4. MongoDB CDC Connector 并行化 Snapshot 改良
5. 后续布局

点击查看直播回放 & 演讲 PDF

一、MongoDB 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 Changelog，便能够在其之上定义成一张动静表，应用 Flink SQL 进行解决。

目前 MongoDB CDC Connector 反对 Exactly-Once 语义，反对全量加增量的订阅，反对从检查点、保留点复原，反对 Snapshot 数据的过滤，反对数据库的 Database、Collection 等元数据的提取，也反对库汇合的正则筛选性能。

二、MongoDB CDC Connector 业务实际

XTransfer 成立于 2017 年，聚焦于 B2B 跨境领取业务，为从事跨境电商进口的中小微企业提供外贸收款以及风控服务。跨境 B 类业务结算场景波及的业务链路很长，从询盘到最终的成交，过程中波及物流条款、领取条款等，须要在每个环节上做好危险管控，以合乎跨境资金交易的监管要求。

以上种种因素对 XTransfer 的数据处理安全性和准确性都提出了更高的要求。在此基础上，XTransfer 基于 Flink 搭建了本人的大数据平台，可能无效保障在跨境 B2B 全链路上的数据可能被无效地采集、加工和计算，并满足了高平安、低提早、高精度的需要。

变更数据采集 CDC 是数据集成的关键环节。在没有应用 Flink CDC 之前，个别应用 Debezium、Canal 等传统 CDC 工具来抽取数据库的变更日志，并将其转发到 Kafka 中，上游读取 Kafka 中的变更日志进行生产。这种架构存在以下痛点：

• 部署组件多，运维老本较高；
• 上游数据生产逻辑须要依据写入端进行适配，存在肯定的开发成本；
• 数据订阅配置较简单，无奈像 Flink CDC 一样仅通过 SQL 语句便定义出一个残缺的数据同步逻辑；
• 难以全副满足全量 + 增量采集，可能须要引入 DataX 等全量采集组件；
• 比拟偏差于对变更数据的采集，对数据的解决过滤能力较为单薄；
• 难以满足异构数据源打宽的场景。

目前咱们的大数据平台次要应用 Flink CDC 来进行变更数据捕捉，它具备如下劣势：

1. 实时数据集成

• 毋庸额定部署 Debezium、Canal、Datax 等组件，运维老本大幅升高；
• 反对丰盛的数据源，也可复用 Flink 既有的 connectors 进行数据采集写入，能够笼罩大多数业务场景；
• 升高了开发难度，仅通过 Flink SQL 就能够定义出残缺的数据集成工作流程；
• 数据处理能力较强，依靠于 Flink 平台弱小的计算能力能够实现流式 ETL 甚至异构数据源的 join、group by 等。

2. 构建实时数仓

• 大幅简化实时数仓的部署难度，通过 Flink CDC 实时采集数据库的变更，并写入 Kafka、Iceberg、Hudi、TiDB 等数据库中，即可应用 Flink 进行深度的数据挖掘和数据处理。
• Flink 的计算引擎能够反对流批一体的计算模式，不必再保护多套计算引擎，能够大幅升高数据的开发成本。

3. 实时风控

• 实时风控以往个别采取往 Kafka 中发业务事件的形式实现，而应用 Flink CDC 之后，能够间接从业务库中捕捉风控事件，而后通过 Flink CDC 来进行简单的事件处理。
• 能够运行模型，以通过 Flink ML、Alink 来丰盛机器学习的能力。最初将这些实时风控的处理后果回落进 Kafka，下达风控指令。

三、MongoDB CDC Connector 生产调优

MongoDB CDC Connector 的应用有如下几点要求：

• 鉴于应用了 Change Streams 的个性来实现 MongoDB CDC Connector，因而要求 MongoDB 的最小可用版本是 3.6，比拟举荐 4.0.8 及以上版本。
• 必须应用集群部署模式。因为订阅 MongoDB 的 Change Streams 要求节点之间可能进行互相复制数据，单机 MongoDB 无奈进行数据的相互拷贝，也没有 Oplog，只有正本集或分片集的状况下才有数据复制机制。
• 须要应用 WireTiger 存储引擎，应用 pv1 复制协定。
• 须要领有 ChangeStream 和 find 用户权限。

应用 MongoDB CDC Connector 时要留神设置 Oplog 的容量和过期工夫。MongoDB oplog 是一个非凡的有容量汇合，容量达到最大值后，会抛弃历史数据。而 Change Streams 通过 resume token 来进行复原，太小的 oplog 容量可能会导致 resume token 对应的 oplog 记录不再存在，即 resume token 过期，进而导致 Change Streams 无奈被复原。

能够应用 replSetResizeOplog 设置 oplog 容量和最短保留工夫，MongoDB 4.4 版本之后也反对设置最小工夫。一般而言，生产环境中倡议 oplog 保留不小于 7 天。

对一些变更较慢的表，倡议在配置中开启心跳事件。变更事件和心跳事件能够同时向前推动 resume token，对于变更较慢的表，能够通过心跳事件来刷新 resume token 防止其过期。

能够通过 heartbeat.interval.ms 设置心跳的距离。

因为只能将 MongoDB 的 Change Streams 转换成 Flink 的 Upsert changelog，它相似于 Upsert Kafka 模式，为了补齐 –U 前置镜像值，会减少一个算子 ChangelogNormalize，而这会带来额定的状态开销。因而在生产环境中比拟举荐应用 RocksDB State Backend。

当默认连贯的参数无奈满足应用需要时，能够通过设置 connection.options 配置项来传递 MongoDB 反对的连贯参数。

比方连贯 MongoDB 的用户创立的数据库不在 admin 中，能够设置参数来指定须要应用哪个数据库来认证以后用户，也能够设置连接池的最大连贯参数等，MongoDB 的连贯字符串默认反对这些参数。

正则匹配多库、多表是 MongoDB CDC Connector 在 2.0 版本之后提供的新性能。须要留神，如果数据库名称应用了正则参数，则须要领有 readAnyDatabase 角色。因为 MongoDB 的 Change Streams 只能在整个集群、数据库以及 collection 粒度上开启。如果须要对整个数据库进行过滤，那么数据库进行正则匹配时只能在整个集群上开启 Change Streams，而后通过 Pipeline 过滤数据库的变更。能够通过在 Ddatabase 和 Collection 两个参数中写入正则表达式进行多库、多表的订阅。

四、MongoDB CDC Connector 并行化 Snapshot 改良

为了减速 Snapshot 的速度，能够应用 Flip-27 引入的 source 来进行并行化革新。首先应用一个 split 枚举器，依据肯定的切分策略，将一个残缺的 Snapshot 工作拆分成若干个子工作，而后调配给多个 split reader 并行做 Snapshot，以此晋升整体工作的运行速度。

然而在 MongoDB 里，大多状况下组件是 ObjectID，其中后面四个字节是 UNIX 形容，两头五个字节是一个随机值，前面三个字节是一个自增量。在雷同形容里插入的文档并不是严格递增的，两头的随机值可能会影响部分的严格递增，但从总体来看，仍然可能满足递增趋势。

因而，不同于 MySQL 的递增组件，MongoDB 并不适宜采纳 offset + limit 的切分策略对其汇合进行简略拆分，须要针对 ObjectID 采纳针对性的切分策略。

最终，咱们采取了以下三种 MongoDB 切分策略：

• Sample 采样分桶：原理是利用 $sample 命令对 collection 进行随机采样，通过均匀文档大小和每个 chunk 的大小来预估须要的分桶数。要求相应汇合的查问权限，其长处是速度较快，实用于数据量大然而没有分片的汇合；毛病是因为应用了抽样预估模式，分桶的后果不能做到相对平均。
• SplitVector 索引切分：SplitVector 是 MongoDB 计算 chunk 决裂点的外部命令，通过拜访指定的索引计算出每个 chunk 的边界。要求领有 SplitVector 权限，其长处是速度快，chunk 后果平均；毛病是对于数据量大且曾经分片的汇合，不如间接读取 config 库中曾经分好的 chunks 元数据。
• Chunks 元数据读取：因为 MongoDB 在 config 数据库会存储分片汇合的理论分片后果，因而能够间接从 config 中读取分片汇合的理论分片后果。要求领有 config 库读取权限，仅限于分片汇合应用。其长处是速度快，毋庸从新计算 chunk 决裂点，chunk 后果平均，默认状况下为 64MB；毛病是不能满足所有场景，仅限分片场景。

上图为 sample 采样分桶示例。左侧是一个残缺的汇合，从残缺的汇合中设定样本数量，而后将整个样本放大，并依据采样当前的样本进行分桶，最终后果就是咱们心愿的 chunks 边界。

sample 命令是 MongoDB 采样的一个内置命令。在样本值小于 5% 的状况下，应用伪随机算法进行采样；样本值大于 5% 的状况下，先应用随机排序，而后抉择前 N 个文档。它的均匀度和耗时次要取决于随机算法和样本的数量，是一种平均水平和切分速度的折中策略，适宜于要求切分速度快，但能够容忍切分后果不太平均的场景。

在理论测试中，sample 采样的平均水平有着不错的体现。

上图为 SplitVector 索引切分示例。左侧是原始汇合，通过 SplitVector 命令指定须要拜访的索引，为 ID 索引。能够设置每个 chunk 的大小，单位为 MB，而后应用 SplitVector 命令拜访索引，并通过索引计算每个块的边界。

它速度快，chunk 后果也很平均，实用于大部分场景。

上图为 config.chuncks 读取示例，即间接读取 MongoDB 曾经分好的 chunks 元数据。在 Config Server 中会存储每个 Shard、其所在机器以及每个 Shard 的边界。对于分片汇合，能够间接在 chunks 中读取它的边界信息，毋庸反复计算这些决裂点，也能够保障每一个 chunk 的读取在单台机器上就能实现，速度极快，在大规模的分片汇合场景下有着很好的体现。

五、后续布局

Flink CDC 的后续布局次要分为以下五个方面：

• 第一，帮助欠缺 Flink CDC 增量 Snapshot 框架；
• 第二，应用 MongoDB CDC 对接 Flink CDC 增量 Snapshot 框架，使其可能反对并行 Snapshot 改良；
• 第三，MongoDB CDC 反对 Flink RawType。对于一些比拟灵便的存储构造提供 RawType 转换，用户能够通过 UDF 的模式对其进行自定义解析；
• 第四，MongoDB CDC 反对从指定地位进行变更数据的采集；
• 第五，MongoDB CDC 稳定性的优化。

问答

Q：MongoDB CDC 提早高吗？是否须要通过就义性能来升高提早？

A：MongoDB CDC 提早不高，在全量采集的时候通过 changelog normalize 可能会对于 CDC 的增量采集造成一些背压，然而这种状况能够通过 MongoDB 并行化革新、减少资源的形式来防止。

Q：默认连贯什么时候无奈满足要求？

A：MongoDB 的用户能够在任何数据库、任何子库中进行创立。如果不是在 admin 的数据库中创立用户，认证的时候须要显示地指定要在哪个数据库中认证用户，也能够设置最大的连贯大小等参数。

Q：MongoDB 目前的 DBlog 反对无锁并发读取吗？

A：DBlog 的无锁并发领有增量快照的能力，然而因为 MongoDB 难以获取以后 changelog 的位点，所以增量快照无奈立即实现，但无锁并发的 Snapshot 行将反对。

更多 Flink 相干技术问题，可扫码退出社区钉钉交换群
第一工夫获取最新技术文章和社区动静，请关注公众号～

流动举荐

阿里云基于 Apache Flink 构建的企业级产品 - 实时计算 Flink 版现开启流动：
99 元试用 实时计算 Flink 版（包年包月、10CU）即有机会取得 Flink 独家定制卫衣；另包 3 个月及以上还有 85 折优惠！
理解流动详情：https://www.aliyun.com/produc…