Flink CDC MongoDB 联合实时数仓的探索实践
一、 在实时数仓的探索
是一款非关系型的文档数据库，支持大规模的数据存储和灵活的存储结构，在 内部有着比较大规模的应用。
另外， 在实时数仓方面也有着积极的探索，除了目前比较流行的基于湖技术的构建实时数仓的方式外，Flink 和 也有着构建轻量级实时数仓的潜力。
1.1 简介
是一种面向文档的非关系型数据库，支持半结构化的数据存储。它还是一种分布式的数据库，提供副本集和分片级两种集群部署模式，具有高可用和水平扩展的能力，适合大规模的数据存储。另外， 在 3.0 版本之后还引入了 特性，支持并简化了数据库的变更订阅。
1.2 常见的实时架构选型
优势包括，数据新鲜度高；数据写入较快；Kafka 的周边组件生态较好。
缺陷包括，中间结果不可查。Kafka 是线性存储，记录了数据的每一次变更，因此如果要得到最新的镜像值，需要遍历所有在 Kafka 中的记录，因此也无法进行比较灵活快速的 OLAP 查询，对于排查问题方面也比较困难；Kafka 的冷热分离还有待实现，不能充分利用一些廉价存储；这套架构一般需要额外维护两套流批架构，对部署开发运维成本会较高。
目前比较流行的数据湖 、Hudi，同时支持了批式读取和流式读取的能力，可以通过 Flink 实现流批一体的计算能力，其次，湖存储在存储上会充分考虑如何利用廉价存储，相对于 Kafka 具有更低的存储成本。
但基于湖存储的实时数仓也有一些缺点，包括部署成本较高，例如需要额外部署一些 OLAP 查询引擎。其次，对于数据权限也需要额外的组件来支持。
本身支持大规模数据集存储，也支持灵活的数据格式； 的部署成本低，组件依赖少，并且有完整的权限控制。相比于其他的实时数仓架构，Flink 和 也有着构建轻量级实时数仓的潜力。这种模式要求 Flink 对 拥有流式读取、批式读取、维表查询和行级写入的能力。
目前全增量一体化流式查询可以通过 Flink CDC 提供，批式读取维表查询写入的功能可以由 FLIP-262 提供。
二、 CDC 的实现原理和使用实践 2.1 CDC
CDC 由 基础架构团队开发，并已贡献给了 Flink CDC 社区。在 Flink CDC 2.1.0 版本中正式引入，支持了全增量一体化的 CDC 读取以及元数据提取的功能；2.1.1 版本中，支持连接未开启认证的 ；2.2.0 版本中，支持正则表达式筛选的功能；2.3.0 版本中，基于增量快照读框架，实现了并行增量快照读的功能。
2.2 特性
CDC 是基于 特性来实现的。 是一个分布式的数据库，在分布式的环境中，集群成员之间一般会进行相互复制，来确保数据的完整性和容错性。与 MySQL 的 比较类似， 也提供了 oplog 来记录数据的操作变更，次要节点之间通过访问主节点的变更日志来进行数据的同步。
我们也可以通过直接遍历 oplog 的方式来获取数据库的变更。但分片集群一般由多个 shard 组成，每个 shard 一般也是一个独立的副本集。在分片上的 oplog 仅包含在它分片范围里的数据，因此我们需要遍历所有 shard 上的 oplog，并把它们根据时间进行排序合并，这显然会比较复杂。
值得一提的是，在 3.6 版本之后，引入了 特性，提供了更简单的 API 来简化数据订阅。
使用 的 API，我们可以屏蔽遍历 oplog 并整合的复杂度，并且支持实例、库、集合等多种级别的订阅方式，以及完整的故障恢复机制。
2.3 的故障恢复
通过 来进行断点恢复， 返回的每条记录都会携带一个 ，每个 都对应了 oplog 中的一条具体记录，表示已经读到的 oplog 的位置。另外，还记录了变更时间以及变更文档主键的信息。通过 、 等方法，将 作为起始参数可以对中断的 进行恢复。
的 是由 编码的一个字符串，它的结构如上图左侧所示。ts 代表数据发生变更的时间，ui 代表发生变更 的 UUID，o2 代表发生变更的文档的主键。详细的 oplog 字段描述可以参考。
上图右侧是一个 oplog 的具体记录，它描述了在 107 结尾主键下的一条记录的一次更改，将 字段修改成了 5.4。值得一提的是 在 6.0 版本中并没有提供变更前和变更后完整的镜像值。这也是我们没有直接采用 oplog 去实现 CDC 的一个原因。
2.4 的演进
在 3.6 版本中正式引入了变更流特性，但仅支持对于单个集合的订阅。在 4.0 版本支持了实例、库级别的订阅，也支持了指定时间戳开启变更流的功能。在 4.0.7 版本引入了 ：
在 4.0 版本之前打开一个变更流后，如果没有新的变更数据产生，那么将不会获取到最新的 。如果此时发生故障，并且尝试使用了比较老旧的 来恢复，可能会降低服务器的性能，因为服务器可能会需要扫描更多的 oplog 的条目。如果 对应的 oplog 被清除了，那么这个变更流将无法进行恢复。
为了解决这个问题， 4.0 提供了 ，标记已经扫描的 oplog 的位置，并且会随时间持续推进。另外，利用这个特性，我们可以比较准确的定位当前 消费的位置，进而实现增量快照读的功能。
在 4.2 版本，可以使用 去处理一些 的事件，在 5.1 版本对 进行了一系列的优化。在 6.0 版本，提供了 前置、后置镜像值完整信息，以及 变更的订阅机制。
2.5 CDC
CDC 的实现原理，是利用了 的特性，将增、删、改等变更事件转换成 Flink 的 类型的变更流。在 Flink SQL 场景下， 会加上 的算子，将 类型的变更流进行标准化。结合 Flink 强大的计算能力，容易实现实时 ETL 甚至异构异构数据源的计算场景。
在 Flink CDC 2.3 版本，依托于增量快照读框架实现了无锁快照读的功能，支持并发快照，大大缩短了快照时间。关于增量快照读的总体流程是如上图所示。为了让 并行化，首先要将完整的数据集切分成多个区块。将这些区块分配给不同的 并行读取，以提升整个 的速度。但 的主键它多为 ，不能按照简单的增加范围的方式去切分，因此对于 的切分策略需要单独去设计。
有以下三种切分策略，这些切分策略参考了 Mongo Spark 项目。
接下来介绍一下增量快照读的过程。对于一个已经切分好的区块，在快照执行前后分别记录当前 的位置。在快照结束之后，根据快照起始、结束的位点范围，对变更流进行回放，最后将快照记录和变更记录按 Key 进行合并，得到完整的结果，避免了重复数据的发送。
在单个 Chunk 的增量读阶段，我们读取了 Chunk 范围内的快照数据以及 Chunk 范围内的增量数据，并将其进行合并。但整体的 的过程可能并没有结束，那么已经完成 的区块，在后边的时间仍然可能会发生变更，因此我们需要对这些变更数据进行补偿。从全局最低的高水位点处开始启动变更流，对于变更时间高于 Chunk 高位点的变更数据进行补偿。当达到全局 最高位点的时候，我们的补偿便可以结束。
接下来介绍一些关于 CDC 的生产建议。
三、FLIP-262 的功能预览
上面我们介绍了 CDC ，可以对 进行增量的 CDC 读取，但如果要在 上构建实时数仓，我们还需要对 进行批量读取、写入以及 的能力。这些功能在 FLIP-262 中进行实现，目前已经发布第一个版本。
3.1 FLIP-262
在并行读取方面， 基于 FLIP-27 新的 API 实现；支持批量读取；支持 。在并行写入方面，基于 FLIP-177 Sink API 实现；支持 写入。在 Table API 方面，实现了 FLIP-95 Table API 使用 Flink SQL 进行读取或写入。
3.2 读取
首先我们在 中插入一些测试数据，然后使用 Flink SQL 定义一张 users 表，通过 语句我们可以得到右边所示的结果。可以发现右边的结果和我们插入的测试数据是一致的。
3.3 写入
首先我们定义一张 users 的结果表，对应 users 的集合。然后我们通过 Flink SQL 的 语句，将上面定义的 users 表集合的数据，读取并写入到 。
最后查询一下我们新定义的这张结果表，它的结果如右边所示。可以发现它的结果和之前源表的结果是一致的，这代表着我们写入一张新的集合是成功的。
3.4 用作维表关联
接下来来演示一下，将上面定义的 user 表作为维表进行 的场景。
首先我们定义一张 的事实表， 作为 Key，对应于我们之前定义的 users 表的主键。然后我们查询 表可以得到右边的结果。
接着定义一张结果表代表打款以后的结果，这个结果表对于 users 是作为维表关联去补充一些区域信息。然后我们通过 Flink SQL 将 事实表和 users 维表进行关联，写入到结果表。然后查询结果表可以得到打宽后 的信息。如右图所示，打宽以后的 在最后一列，这说明我们的 是成功的。
四、总结和展望 4.1 总结
至此，Flink 联合 的实时数仓架构便可以实现，在建设实时数仓时多了一份选择。如图所示，通过 CDC 完成整套流式链路，辅助 进行数据打宽。通过 完成一整套批式链路，最后将计算的中间结果通过 Sink 进行存储，那么整套实时数仓的架构便得以实现。
4.2 存在的问题
目前还存在着以下问题：
4.3 未来规划
在 CDC 方面，我们需要：
在 方面，我们需要：
