快速实践: 通过 Flink CDC 一键整库同步 MongoDB 到 Paimon

原创
2023/12/26 16:24
阅读数 41

导言

MongoDB 是一个比较成熟的文档数据库,在业务场景中,通常需要采集 MongoDB 的数据到数据仓库或数据湖中,面向分析场景使用。

Flink MongoDB CDC 是 Flink CDC 社区提供的一个用于捕获变更数据(Change Data Capturing)的 Flink 连接器,可连接到 MongoDB 数据库和集合,并捕获其中的文档增加、更新、替换、删除等变更操作。

Apache Paimon (incubating) 是一项流式数据湖存储技术,可以为用户提供高吞吐、低延迟的数据摄入、流式订阅以及实时查询能力。

Paimon CDC

https://paimon.apache.org/docs/master/cdc-ingestion/overview/

Paimon CDC 是整合了 Flink CDC、Kafka、Paimon 的入湖工具,帮助你更好更方便的完成一键入湖。

你可以通过 Flink SQL 或者 Flink DataStream API 将 Flink CDC 数据写入 Paimon 中,也可以通过Paimon 提供的 CDC 工具来完成入湖。那这两种方式有什么区别呢?

1

上图是使用 Flink SQL 来完成入湖,简单,但是当源表添加新列后,同步作业不会同步新的列,下游 Paimon 表也不会增加新列。

2

上图是使用 Paimon CDC 工具来同步数据,可以看到,当源表发生列的新增后,流作业会自动新增列的同步,并传导到下游的 Paimon 表中,完成 Schema Evolution 的同步。

另外 Paimon CDC 工具也提供了整库同步:

3

整库同步可以帮助你:

  1. 一个作业同步多张表,以低成本的方式同步大量小表

  2. 作业里同时自动进行 Schema Evolution

  3. 新表将会被自动进行同步,你不用重启作业,全自动完成

Demo 说明

你可以跟随 Demo 步骤体验 Paimon CDC 的全自动同步之旅,Demo 展示同步 Mongo DB 的数据到Paimon 中,如下图。

4

以下的 Demo 使用 Flink 来完成入湖,使用 Spark SQL 来查询,当然你可以使用 Flink SQL 来查询,或者使用其它计算引擎,包括 Trino、Presto、StarRocks、Doris 、Hive 等等。

Demo 准备

步骤一:

首先下载 MongoDB Community Server,免费版,不用交钱。

https://www.mongodb.com/try/download/community

启动 MongoDB Server:

mkdir /tmp/mongodata ./mongod --replSet rs0 --dbpath /tmp/mongodata

注意:这里开启了replSet,详见 MongoDB 文档,只有开启了 replSet 的库才会产生 changelog,也就才会被 Flink Mongo CDC 可以增量读取 CDC 数据。

步骤二:

下载 MongoDB Shell:

https://www.mongodb.com/try/download/shell

并启动:

./mongosh

另外需要初始化 replSet,否者 MongoDB Server 会一直报错。

rs.initiate()

步骤三:

下载 Flink,请到官网下载最新 Flink:

https://www.apache.org/dyn/closer.lua/flink/flink-1.18.0/flink-1.18.0-bin-scala_2.12.tgz

并依次下载以下 Jars 到 Flink 的 lib 目录中:

paimon-flink-1.18-0.6-*.jar,paimon-flink 集成 Jar:

https://repository.apache.org/snapshots/org/apache/paimon/paimon-flink-1.18/0.6-SNAPSHOT/

flink-shaded-hadoop-*.jar,Paimon 需要 hadoop 相关依赖:

https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar

flink-sql-connector-mongodb-cdc-*.jar:

https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mongodb-cdc/2.4.2/flink-sql-connector-mongodb-cdc-2.4.2.jar

在 flink/conf/flink-conf.yaml 文件中设置 checkpoint 间隔:

execution.checkpointing.interval: 10 s

生产中不推荐使用此间隔,太快会产生大量文件导致 Cost 上升,一般推荐的 Checkpoint 间隔是 1 - 5 分钟。

启动 Flink 集群:

./bin/start-cluster.sh

启动 Flink 同步任务:

./bin/flink run lib/paimon-flink-action-0.6-*.jar   mongodb-sync-database    --warehouse /tmp/warehouse1   --database test   --mongodb-conf hosts=127.0.0.1:27017   --mongodb-conf database=test   --table-conf bucket=1

参数说明:

  1. Warehouse 指定 paimon 所在文件系统目录,如你有 HDFS 集群或者对象存储,可以替换成你的目录。

  2. MongoDB 相关配置,如有密码,请填写密码。

  3. 最后指定 bucket 个数,目前整库同步只支持了固定 Bucket 的表,如有特殊需求,可以修改个别表的 Bucket 个数。

5

可以看到,作业已成功启动,拓扑主要包含三个节点:

  1. Source:Flink MongoDB CDC Source,并完成 Schema Evolution 和自动加表。

  2. CDC MultiplexWriter:复杂多个表的 Paimon 表 Writer,自动动态加表。

  3. Multiplex Global Committer:两阶段提交的文件提交节点。

Writer 和 Committer 都有可能成为瓶颈,Writer 和 Committer 的并发都可以通过 Flink 的配置影响。

你可以考虑打开全异步模式来避免 Writer 的 Compaction 瓶颈:

https://paimon.apache.org/docs/master/maintenance/write-performance/#asynchronous-compaction

步骤四:

下载 Spark,请到官网下载最新版本:

https://spark.apache.org/downloads.html

下载 Paimon Spark 集成 Jar:

https://repository.apache.org/content/groups/snapshots/org/apache/paimon/paimon-spark-3.5/0.6-SNAPSHOT/

启动 Spark SQL:

./bin/spark-sql   --conf spark.sql.catalog.paimon=org.apache.paimon.spark.SparkCatalog  --conf spark.sql.catalog.paimon.warehouse=file:/tmp/warehouse1

使用 Paimon Catalog,指定 Database:

USE paimon;USE rs0;

Demo 开始

步骤一:

我们首先测试下写入的数据可以被成功读取到。

我们先给 MongoDB 插入一条数据:

db.orders.insertOne({id: 1, price: 5})

然后我们在 Spark SQL 里查询:

6

可以看到这条数据被同步到 Paimon 里,并且可以看到 orders 表的 Schema 里多了一列 “_id”,这列是MongoDB 自动生成的隐含的主键。

步骤二:

我们再来看看更新是如何被同步的。

在 Mongo Shell 里更新下数据:

db.orders.update({id: 1}, {$set: { price: 8 }})

Spark 里查询:

7

数据的 price 被更新为 8

步骤三:

我们再来看看添加字段的同步情况。

在 Mongo Shell 里新插入一条数据,多了一列:

db.orders.insertOne({id: 2, price: 6, desc: “haha”})

Spark 里查询:

8

可以看到,Paimon 对应的表里已经新增了一列,查询数据显示,老的数据默认值为 NULL。

步骤四:

我们再来看看新增表的同步情况。

在 Mongo Shell 里新插入一张表的数据:

db.brands.insertOne({id: 1, brand: “NBA”})

Spark 里查询:

Paimon 里已经自动多出来一张表,数据也被同步过来。

总结

通过上面的操作你感受到了吗,通过 Paimon CDC 的入湖程序可以让你全自动的同步业务数据库到 Paimon 里,数据、Schema Evolution、新增表,全部被自动完成,你只用管好这一个 Flink 作业即可。这套入湖程序已经被部署到各行各业,各个公司里,给业务数据带来非常方便的镜像到湖存储里面的能力。

更有其它数据源等你来体验:Mysql、Kafka、MongoDB、Pulsar、PostgresSQL。

Paimon 的长期使命包括:

  • 极致易用性、高性能的数据入湖,方便的湖存储管理,丰富生态的查询。
  • 方便的数据流读,与 Flink 生态的良好集成,给业务带来1分钟新鲜度的数据。
  • 加强的 Append 数据处理,时间旅行、数据排序带来高效的查询,升级 Hive 数仓。

关于 Paimon

  • 微信公众号:Apache Paimon ,了解行业实践与最新动态
  • 官网:https://paimon.apache.org/ 查询文档和关注项目


更多内容


活动推荐

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

展开阅读全文
加载中
点击引领话题📣 发布并加入讨论🔥
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部