Flink CDC & MongoDB 联合实时数仓的探索实践

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: XTransfer 技术专家, Flink CDC Maintainer 孙家宝,在 Flink Forward Asia 2022 数据集成专场的分享。

摘要:本文整理自 XTransfer 技术专家, Flink CDC Maintainer 孙家宝,在 Flink Forward Asia 2022 数据集成专场的分享。本篇内容主要分为四个部分:

  1. MongoDB 在实时数仓的探索

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

  3. FLIP-262 MongoDB Connector 的功能预览

  4. 总结和展望

点击查看原文视频 & 演讲PPT

一、MongoDB 在实时数仓的探索

MongoDB 是一款非关系型的文档数据库,支持大规模的数据存储和灵活的存储结构,在 XTransfer 内部有着比较大规模的应用。

另外,XTransfer 在实时数仓方面也有着积极的探索,除了目前比较流行的基于湖技术的构建实时数仓的方式外,Flink 和 MongoDB 也有着构建轻量级实时数仓的潜力。

1.1 MongoDB 简介

1

MongoDB 是一种面向文档的非关系型数据库,支持半结构化的数据存储。它还是一种分布式的数据库,提供副本集和分片级两种集群部署模式,具有高可用和水平扩展的能力,适合大规模的数据存储。另外,MongoDB 在 3.0 版本之后还引入了 Change Streams 特性,支持并简化了数据库的变更订阅。

1.2 常见的实时架构选型

2

  • Flink 和 Kafka 纯实时链路的实时数仓。

优势包括,数据新鲜度高;数据写入较快;Kafka 的周边组件生态较好。

缺陷包括,中间结果不可查。Kafka 是线性存储,记录了数据的每一次变更,因此如果要得到最新的镜像值,需要遍历所有在 Kafka 中的记录,因此也无法进行比较灵活快速的 OLAP 查询,对于排查问题方面也比较困难;Kafka 的冷热分离还有待实现,不能充分利用一些廉价存储;这套架构一般需要额外维护两套流批架构,对部署开发运维成本会较高。

3

  • 基于湖存储的实时数仓架构。

目前比较流行的数据湖 Iceberg、Hudi,同时支持了批式读取和流式读取的能力,可以通过 Flink 实现流批一体的计算能力,其次,湖存储在存储上会充分考虑如何利用廉价存储,相对于 Kafka 具有更低的存储成本。

但基于湖存储的实时数仓也有一些缺点,包括部署成本较高,例如需要额外部署一些 OLAP 查询引擎。其次,对于数据权限也需要额外的组件来支持。

4

  • 基于 MongoDB 的实时数仓。

MongoDB 本身支持大规模数据集存储,也支持灵活的数据格式;MongoDB 的部署成本低,组件依赖少,并且有完整的权限控制。相比于其他的实时数仓架构,Flink 和 MongoDB 也有着构建轻量级实时数仓的潜力。这种模式要求 Flink 对 MongoDB 拥有流式读取、批式读取、维表查询和行级写入的能力。

目前全增量一体化流式查询可以通过 Flink CDC MongoDB Connector 提供,批式读取维表查询写入的功能可以由 FLIP-262 MongoDB Connector 提供。

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

2.1 MongoDB CDC Connector

5

MongoDB CDC Connector 由 XTransfer 基础架构团队开发,并已贡献给了 Flink CDC 社区。在 Flink CDC 2.1.0 版本中正式引入,支持了全增量一体化的 CDC 读取以及元数据提取的功能;2.1.1 版本中,支持连接未开启认证的 MongoDB;2.2.0 版本中,支持正则表达式筛选的功能;2.3.0 版本中,基于增量快照读框架,实现了并行增量快照读的功能。

2.2 Change Streams 特性

6

MongoDB CDC Connector 是基于 MongoDB Change Streams 特性来实现的。MongoDB 是一个分布式的数据库,在分布式的环境中,集群成员之间一般会进行相互复制,来确保数据的完整性和容错性。与 MySQL 的 Binlog 比较类似,MongoDB 也提供了 oplog 来记录数据的操作变更,次要节点之间通过访问主节点的变更日志来进行数据的同步。

我们也可以通过直接遍历 MongoDB oplog 的方式来获取数据库的变更。但分片集群一般由多个 shard 组成,每个 shard 一般也是一个独立的副本集。在分片上的 oplog 仅包含在它分片范围里的数据,因此我们需要遍历所有 shard 上的 oplog,并把它们根据时间进行排序合并,这显然会比较复杂。

值得一提的是,在 MongoDB 3.6 版本之后,引入了 Change Streams 特性,提供了更简单的 API 来简化数据订阅。

7

使用 Change Streams 的 API,我们可以屏蔽遍历 oplog 并整合的复杂度,并且支持实例、库、集合等多种级别的订阅方式,以及完整的故障恢复机制。

2.3 Change Streams 的故障恢复

8

MongoDB 通过 ResumeToken 来进行断点恢复,Change Streams 返回的每条记录都会携带一个 ResumeToken,每个 ResumeToken 都对应了 oplog 中的一条具体记录,表示已经读到的 oplog 的位置。另外,还记录了变更时间以及变更文档主键的信息。通过 ResumeAfter、startAfter 等方法,将 ResumeToken 作为起始参数可以对中断的 Change Streams 进行恢复。

Change Streams 的 ResumeToken 是由 MongoDB KeyStream 编码的一个字符串,它的结构如上图左侧所示。ts 代表数据发生变更的时间,ui 代表发生变更 collection 的 UUID,o2 代表发生变更的文档的主键。详细的 oplog 字段描述可以参考 oplog_entry

上图右侧是一个 oplog 的具体记录,它描述了在 107 结尾主键下的一条记录的一次更改,将 weight 字段修改成了 5.4。值得一提的是 MongoDB 在 6.0 版本中并没有提供变更前和变更后完整的镜像值。这也是我们没有直接采用 oplog 去实现 MongoDB CDC Connector 的一个原因。

2.4 Change Streams 的演进

9

MongoDB 在 3.6 版本中正式引入了变更流特性,但仅支持对于单个集合的订阅。在 4.0 版本支持了实例、库级别的订阅,也支持了指定时间戳开启变更流的功能。在 4.0.7 版本引入了 postBatchResumeToken:

在 4.0 版本之前打开一个变更流后,如果没有新的变更数据产生,那么将不会获取到最新的 ResumeToken。如果此时发生故障,并且尝试使用了比较老旧的 ResumeToken 来恢复,可能会降低服务器的性能,因为服务器可能会需要扫描更多的 oplog 的条目。如果 ResumeToken 对应的 oplog 被清除了,那么这个变更流将无法进行恢复。

为了解决这个问题,MongoDB 4.0 提供了 postBatchResumeToken,标记已经扫描的 oplog 的位置,并且会随时间持续推进。另外,利用这个特性,我们可以比较准确的定位当前 Change Streams 消费的位置,进而实现增量快照读的功能。

在 MongoDB 4.2 版本,可以使用 startAfter 去处理一些 invalid 的事件,在 MongoDB 5.1 版本对 Change Streams 进行了一系列的优化。在 MongoDB 6.0 版本,提供了 Change Streams 前置、后置镜像值完整信息,以及 Schema 变更的订阅机制。

2.5 MongoDB CDC Connector

10

MongoDB CDC Connector 的实现原理,是利用了 Change Streams 的特性,将增、删、改等变更事件转换成 Flink 的 upsert 类型的变更流。在 Flink SQL 场景下,Planner 会加上 Changelog Normalize 的算子,将 upsert 类型的变更流进行标准化。结合 Flink 强大的计算能力,容易实现实时 ETL 甚至异构异构数据源的计算场景。

11

在 Flink CDC 2.3 版本,依托于增量快照读框架实现了无锁快照读的功能,支持并发快照,大大缩短了快照时间。关于增量快照读的总体流程是如上图所示。为了让 snapshot 并行化,首先要将完整的数据集切分成多个区块。将这些区块分配给不同的 Source Reader 并行读取,以提升整个 snapshot 的速度。但 MongoDB 的主键它多为 ObjectId,不能按照简单的增加范围的方式去切分,因此对于 MongoDB 的切分策略需要单独去设计。

12

MongoDB 有以下三种切分策略,这些切分策略参考了 Mongo Spark 项目。

  • 第一种切分策略使用了 Sample 命令对集合进行随机采样,再通过文档的平均大小计算出分桶数量。然后将采样数据分配到各个桶中,构成 Chunk 的边界。优点是速度快,适用于数据量大且未分片的集合。缺点是采用抽样预估,Chunk 的大小不能做到绝对均匀。

  • 第二种切分策略使用了 SplitVector 命令。SplitVector 是 MongoDB 分片式计算分裂点的内部命令,通过访问指定索引来计算每个节点 Chunk 的边界。优点是速度快,并且 Chunk 的大小均匀,但它额外需要 SplitVector 命令的执行权限。

  • 第三种切分策略针对于分片集合,对于已经分片好的集合,我们不用重新计算它的分片结果,可以直接读取 MongoDB 已经分片好的结果作为 Check 的边界。优点是速度快,Chunk 的大小均匀,但是 Chunk 的大小无法调节,依赖于 MongoDB 自身对于每个分片的配置,默认大小为 64mb,另外它额外需要对 config 库的读取权限。

13

接下来介绍一下增量快照读的过程。对于一个已经切分好的区块,在快照执行前后分别记录当前 Change Streams 的位置。在快照结束之后,根据快照起始、结束的位点范围,对变更流进行回放,最后将快照记录和变更记录按 Key 进行合并,得到完整的结果,避免了重复数据的发送。

14

在单个 Chunk 的增量读阶段,我们读取了 Chunk 范围内的快照数据以及 Chunk 范围内的增量数据,并将其进行合并。但整体的 snapshot 的过程可能并没有结束,那么已经完成 snapshot 的区块,在后边的时间仍然可能会发生变更,因此我们需要对这些变更数据进行补偿。从全局最低的高水位点处开始启动变更流,对于变更时间高于 Chunk 高位点的变更数据进行补偿。当达到全局 snapshot 最高位点的时候,我们的补偿便可以结束。

15

接下来介绍一些关于 MongoDB CDC Connector 的生产建议。

  • 第一,使用增量快照特性,MongoDB 的最小可用版本在 4.0 版本。因为在 4.0 版本之前,没有发生变更时无法获取 ResumeToken,且也不能够从指定的时间点进行启动,因此难以实现增量快照特性。

    在 MongoDB 4.0.7 版本之后,引入了 postBatchResumeToken,可以比较容易的获取当前 Change Streams 的位置,因此比较推荐的版本在 4.0.7 以上。

16

  • 第二,控制文档的大小不要超过 8mb,因为 MongoDB 对单条文档有 16 mb 的限制,变更文档因为包含一些额外的信息,比如修改的字段是哪些等等,即使原文档没有超过 16mb,变更文档也会超过 16mb 的限制,从而导致 Change Streams 异常终止。这个应该属于 MongoDB Change Streams 的一个缺陷。

    关于 MongoDB 的变更文档可以超过 16mb 的限制,已在 MongoDB 的 issue 中进行推进。

17

  • 第三,在 MongoDB 中分片件其实在开启事务之后允许被修改。但修改分片键可能会引起分片的频繁移动,引起额外的性能开销。另外,修改分片键还可能导致 Update Lookup 功能失效,在 CDC 的场景中可能会导致结果的不一致。

三、FLIP-262 MongoDB Connector 的功能预览

上面我们介绍了 MongoDB CDC Connector,可以对 MongoDB 进行增量的 CDC 读取,但如果要在 MongoDB 上构建实时数仓,我们还需要对 MongoDB 进行批量读取、写入以及 Lookup 的能力。这些功能在 FLIP-262 MongoDB Connector 中进行实现,目前已经发布第一个版本。

3.1 FLIP-262 Introduce MongoDB Connector

18

在并行读取方面,MongoDB Connector 基于 FLIP-27 新的 Source API 实现;支持批量读取;支持 Lookup。在并行写入方面,基于 FLIP-177 Sink API 实现;支持 Upsert 写入。在 Table API 方面,实现了 FLIP-95 Table API 使用 Flink SQL 进行读取或写入。

3.2 读取 MongoDB

19

首先我们在 MongoDB 中插入一些测试数据,然后使用 Flink SQL 定义一张 users 表,通过 select 语句我们可以得到右边所示的结果。可以发现右边的结果和我们插入的测试数据是一致的。

3.3 写入 MongoDB

20

首先我们定义一张 users snapshot 的结果表,对应 MongoDB users snapshot 的集合。然后我们通过 Flink SQL 的 insert 语句,将上面定义的 users 表集合的数据,读取并写入到 MongoDB。

最后查询一下我们新定义的这张结果表,它的结果如右边所示。可以发现它的结果和之前源表的结果是一致的,这代表着我们写入一张新的集合是成功的。

3.4 用作维表关联

21

接下来来演示一下,将上面定义的 user 表作为维表进行 Lookup 的场景。

首先我们定义一张 pageviews 的事实表,user_id 作为 Lookup Key,对应于我们之前定义的 users 表的主键。然后我们查询 pageviews 表可以得到右边的结果。

22

接着定义一张结果表代表打款以后的结果,这个结果表对于 users 是作为维表关联去补充一些区域信息。然后我们通过 Flink SQL 将 pageviews 事实表和 users 维表进行关联,写入到结果表。然后查询结果表可以得到打宽后 user_region 的信息。如右图所示,打宽以后的 user_region 在最后一列,这说明我们的 Lookup 是成功的。

四、总结和展望

4.1 总结

23

至此,Flink 联合 MongoDB 的实时数仓架构便可以实现,在建设实时数仓时多了一份选择。如图所示,通过 CDC Connector 完成整套流式链路,辅助 Lookup 进行数据打宽。通过 Source Connector 完成一整套批式链路,最后将计算的中间结果通过 Sink Connector 进行存储,那么整套实时数仓的架构便得以实现。

4.2 存在的问题

目前还存在着以下问题:

  • Changelog Normalize 是一个有状态的算子,它会一些额外的状态开销。

  • Update Lookup 的完整状态提取,也需要一定的查询开销。

  • 在 MongoDB 中的文档大小有 16mb 限制。如果有一些很大的单条数据,那么可能并不适合采用这种架构。

4.3 未来规划

在 MongoDB CDC Connector 方面,我们需要:

  • 支持 MongoDB 6.0 版本。

  • 支持指定时间点启动。

  • 推进推进 Changelog Normalize 优化。

在 MongoDB Connector 方面,我们需要:

  • 支持谓词下推。

  • 支持 AsyncLookup。

  • 支持 AsyncSink。

点击查看原文视频 & 演讲PPT


更多内容

img


活动推荐

阿里云基于 Apache Flink 构建的企业级产品-实时计算 Flink 版现开启活动:
0 元试用 实时计算 Flink 版(5000CU*小时,3 个月内)
了解活动详情:https://free.aliyun.com/?pipCode=sc

image.png

相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓,并通过Hologres对接BI分析工具(以DataV为例),实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题?实时计算 Flink 极客训练营产品、技术专家齐上阵,从开源 Flink功能介绍到实时计算 Flink 优势详解,现场实操,5天即可上手! 欢迎开通实时计算 Flink 版: https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍: Flink Forward 是由 Apache 官方授权,Apache Flink Community China 支持的会议,通过参会不仅可以了解到 Flink 社区的最新动态和发展计划,还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验,是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与,一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况,Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
相关文章
|
2月前
|
SQL 分布式计算 DataWorks
破界·融合·进化:解码DataWorks与Hologres的湖仓一体实践
基于阿里云DataWorks与实时数仓Hologres,提供统一的大数据开发治理平台与全链路实时分析能力。DataWorks支持多行业数据集成与管理,Hologres实现海量数据的实时写入与高性能查询分析,二者深度融合,助力企业构建高效、实时的数据驱动决策体系,加速数字化升级。
|
13天前
|
SQL 存储 运维
Apache Doris 在菜鸟的大规模湖仓业务场景落地实践
本文介绍了 Apache Doris 在菜鸟的大规模落地的实践经验,菜鸟为什么选择 Doris,以及 Doris 如何在菜鸟从 0 开始,一步步的验证、落地,到如今上万核的规模,服务于各个业务线,Doris 已然成为菜鸟 OLAP 数据分析的最优选型。
94 2
Apache Doris 在菜鸟的大规模湖仓业务场景落地实践
|
5月前
|
存储 SQL 运维
中国联通网络资源湖仓一体应用实践
本文分享了中国联通技术专家李晓昱在Flink Forward Asia 2024上的演讲,介绍如何借助Flink+Paimon湖仓一体架构解决传统数仓处理百亿级数据的瓶颈。内容涵盖网络资源中心概况、现有挑战、新架构设计及实施效果。新方案实现了数据一致性100%,同步延迟从3小时降至3分钟,存储成本降低50%,为通信行业提供了高效的数据管理范例。未来将深化流式数仓与智能运维融合,推动数字化升级。
199 0
中国联通网络资源湖仓一体应用实践
|
4月前
|
数据采集 SQL canal
Amoro + Flink CDC 数据融合入湖新体验
本文总结了货拉拉高级大数据开发工程师陈政羽在Flink Forward Asia 2024上的分享,聚焦Flink CDC在货拉拉的应用与优化。内容涵盖CDC应用现状、数据入湖新体验、入湖优化及未来规划。文中详细分析了CDC在多业务场景中的实践,包括数据采集平台化、稳定性建设,以及面临的文件碎片化、Schema演进等挑战。同时介绍了基于Apache Amoro的湖仓融合架构,通过自优化服务解决小文件问题,提升数据新鲜度与读写平衡。未来将深化Paimon与Amoro的结合,打造更高效的入湖生态与自动化优化方案。
217 1
Amoro + Flink CDC 数据融合入湖新体验
|
4月前
|
SQL 关系型数据库 MySQL
Flink CDC 3.4 发布, 优化高频 DDL 处理,支持 Batch 模式,新增 Iceberg 支持
Apache Flink CDC 3.4.0 版本正式发布!经过4个月的开发,此版本强化了对高频表结构变更的支持,新增 batch 执行模式和 Apache Iceberg Sink 连接器,可将数据库数据全增量实时写入 Iceberg 数据湖。51位贡献者完成了259次代码提交,优化了 MySQL、MongoDB 等连接器,并修复多个缺陷。未来 3.5 版本将聚焦脏数据处理、数据限流等能力及 AI 生态对接。欢迎下载体验并提出反馈!
731 1
Flink CDC 3.4 发布, 优化高频 DDL 处理,支持 Batch 模式,新增 Iceberg 支持
|
5月前
|
存储 消息中间件 Java
抖音集团电商流量实时数仓建设实践
本文基于抖音集团电商数据工程师姚遥在Flink Forward Asia 2024的分享,围绕电商流量数据处理展开。内容涵盖业务挑战、电商流量建模架构、流批一体实践、大流量任务调优及总结展望五个部分。通过数据建模与优化,实现效率、质量、成本和稳定性全面提升,数据质量达99%以上,任务性能提升70%。未来将聚焦自动化、低代码化与成本优化,探索更高效的流批一体化方案。
320 12
抖音集团电商流量实时数仓建设实践
|
3月前
|
消息中间件 SQL 关系型数据库
Flink CDC + Kafka 加速业务实时化
Flink CDC 是一种支持流批一体的分布式数据集成工具,通过 YAML 配置实现数据传输过程中的路由与转换操作。它已从单一数据源的 CDC 数据流发展为完整的数据同步解决方案,支持 MySQL、Kafka 等多种数据源和目标端(如 Delta Lake、Iceberg)。其核心功能包括多样化数据输入链路、Schema Evolution、Transform 和 Routing 模块,以及丰富的监控指标。相比传统 SQL 和 DataStream 作业,Flink CDC 提供更灵活的 Schema 变更控制和原始 binlog 同步能力。
|
2月前
|
NoSQL MongoDB 数据库
数据库数据恢复—MongoDB数据库数据恢复案例
MongoDB数据库数据恢复环境: 一台操作系统为Windows Server的虚拟机上部署MongoDB数据库。 MongoDB数据库故障: 工作人员在MongoDB服务仍然开启的情况下将MongoDB数据库文件拷贝到其他分区,数据复制完成后将MongoDB数据库原先所在的分区进行了格式化操作。 结果发现拷贝过去的数据无法使用。管理员又将数据拷贝回原始分区,MongoDB服务仍然无法使用,报错“Windows无法启动MongoDB服务(位于 本地计算机 上)错误1067:进程意外终止。”
|
2月前
|
缓存 NoSQL Linux
在CentOS 7系统中彻底移除MongoDB数据库的步骤
以上步骤完成后,MongoDB应该会从您的CentOS 7系统中被彻底移除。在执行上述操作前,请确保已经备份好所有重要数据以防丢失。这些步骤操作需要一些基本的Linux系统管理知识,若您对某一步骤不是非常清楚,请先进行必要的学习或咨询专业人士。在执行系统级操作时,推荐在实施前创建系统快照或备份,以便在出现问题时能够恢复到原先的状态。
205 79

相关产品

  • 实时计算 Flink版
  • 推荐镜像

    更多