Flink MongoDB CDC 实时采集数据到 Kafka 实施方案

适用版本：Apache Flink 1.17.1 Standalone 集群
部署方式：运行时加载 JAR，无需重启集群
更新时间：2026-07-06

一、方案概述

本方案基于 Apache Flink 1.17.1Standalone 本地集群，通过flink-sql-connector-mongodb-cdc连接器实时捕获 MongoDB 的数据变更（Change Data Capture），再经flink-sql-connector-kafka连接器将变更数据写入 Kafka，实现 MongoDB 数据的实时采集与分发。

核心约束：Flink 集群上已有其他业务作业在运行，不能重启集群。Connector JAR 包存放在/data/flink/jars目录，通过 SQL Client 的-l参数在运行时动态加载，无需放入$FLINK_HOME/lib/。

核心技术栈

数据链路

MongoDB (Change Stream) → Flink CDC Source → Flink SQL 处理 → Kafka Sink → 下游消费

免重启加载原理

本方案通过 Flink SQL Client 的嵌入式模式（embedded）提交作业，JAR 包通过-l参数在运行时动态加载到 ClassPath，无需放入$FLINK_HOME/lib/目录，不影响集群中正在运行的其他业务作业。

$FLINK_HOME/bin/sql-client.sh embedded\-l/data/flink/jars\-f/data/flink-cdc-jobs/sql/cdc_to_kafka.sql

二、环境要求

2.1 MongoDB 前置条件

1. MongoDB 必须为副本集（Replica Set）或分片集群（Sharded Cluster），单机模式不支持 Change Stream。
2. MongoDB 版本不低于 4.0（Change Stream 自 3.6 引入，但 4.0+ 稳定性更高）。
3. Flink 集群所有节点（Master + Worker）都必须能网络连通 MongoDB 各节点。CDC 连接器在全量快照阶段需要 JobManager 直连 MongoDB 获取分片元数据，仅开通 Worker 节点网络会导致 Master 节点MongoTimeoutException超时报错。

实际环境连接信息：

验证副本集状态与连通性：

# 测试连通性mongosh"mongodb://flink_user:<your_password>@<mongo-node-1>:27017,<mongo-node-2>:27017/your_database?authSource=admin"

// 登录 mongo shell 后验证副本集状态rs.status()

若为单机模式，需先初始化为单节点副本集：

rs.initiate({_id:"rs0",members:[{_id:0,host:"<mongo-node-1>:27017"}]})

2.2 Kafka 前置条件

1. Kafka 集群版本不低于 2.4。
2. 创建目标 Topic（4 分区、2 副本）：

# 创建 Topickafka-topics.sh --bootstrap-server<kafka-broker-1>:9092\--create--topiccdc_sync_topic\--partitions4--replication-factor2# 验证 Topic 已创建kafka-topics.sh --bootstrap-server<kafka-broker-1>:9092\--describe--topiccdc_sync_topic# 删除 Topic（如需重建）kafka-topics.sh --bootstrap-server<kafka-broker-1>:9092\--delete--topiccdc_sync_topic

3. Flink 集群所在机器能访问 Kafka 各 Broker。

实际环境连接信息：

验证连通性：

# 测试 Broker 连通性telnet<kafka-broker-1>6667telnet<kafka-broker-2>6667telnet<kafka-broker-3>6667telnet<kafka-broker-4>6667# 列出 Topickafka-topics.sh --bootstrap-server<kafka-broker-1>:9092--list# 消费验证（--max-messages 限制条数，避免数据量大导致终端卡顿）kafka-console-consumer.sh\--bootstrap-server<kafka-broker-1>:9092\--topiccdc_sync_topic\--max-messages10\--from-beginning

注意：Kafka 集群 Broker 列表为<kafka-broker-1>:9092,<kafka-broker-2>:9092,<kafka-broker-3>:9092,<kafka-broker-4>:9092，共 4 个节点。Flink Kafka Connector 的properties.bootstrap.servers参数中需按此完整列表填写。

2.3 Flink 环境要求

• Flink 版本1.17.1，解压即可使用。
• 确认$FLINK_HOME/bin/sql-client.sh脚本可执行。
• 关键约束：集群不可重启，JAR 包存放于/data/flink/jars，不可放入$FLINK_HOME/lib/。
• SQL Client embedded 模式会启动独立 MiniCluster，与已有 Standalone 集群互不影响。

# 验证 Flink 安装$FLINK_HOME/bin/sql-client.sh--help

三、JAR 包下载（无需放入 lib 目录）

3.1 必需 JAR 包清单

3.2 下载链接

(1) flink-sql-connector-kafka

• Maven Central 页面：mvnrepository.com/artifact/org.apache.flink/flink-sql-connector-kafka/1.17.1
• 直接下载：

https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/1.17.1/flink-sql-connector-kafka-1.17.1.jar

(2) flink-sql-connector-mongodb-cdc

• Maven Central 页面：mvnrepository.com/artifact/com.ververica/flink-sql-connector-mongodb-cdc/2.4.0
• 直接下载：

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

版本说明：Flink CDC 2.4.x 系列官方声明兼容 Flink 1.13 ~ 1.17，对于 Flink 1.17.1 集群可直接使用2.4.0（2.4 线最新稳定版）。CDC 3.x 系列主要面向 Flink 1.18+，不在本次选用范围。

3.3 目录结构

创建作业目录，SQL 脚本和运行日志统一管理：

mkdir-p/data/flink-cdc-jobs/sqlmkdir-p/data/flink-cdc-jobs/logsmkdir-p/data/flink-cdc-jobs/checkpoint

最终目录结构：

/data/ ├── flink/lib/ # JAR 包（已有，用于 -l 加载） │ ├── flink-sql-connector-kafka-1.17.1.jar │ └── flink-sql-connector-mongodb-cdc-2.4.0.jar └── flink-sql-cdc/ # SQL 作业目录（新建） ├── sql/ # SQL 脚本 ├── logs/ # 运行日志 └── checkpoint/ # Checkpoint 数据

JAR 包若尚未下载，执行：

wget-nc-P/data/flink/jars\https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/1.17.1/flink-sql-connector-kafka-1.17.1.jarwget-nc-P/data/flink/jars\https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mongodb-cdc/2.4.0/flink-sql-connector-mongodb-cdc-2.4.0.jarls-lh/data/flink/jars/flink-sql-connector-*.jar

3.4 运行时加载说明

JAR 包通过 SQL Client 的-l（--library）参数在运行时动态加载。SQL Client 会扫描指定目录下的所有 JAR 文件并追加到 ClassPath，从而使得mongodb-cdc和kafka连接器在 SQL 执行时可用。整个过程无需将 JAR 放入$FLINK_HOME/lib/，也不需要重启 Flink 集群。

# -l 指定 JAR 所在目录，SQL Client 启动时自动加载该目录下所有 JAR$FLINK_HOME/bin/sql-client.sh embedded\-l/data/flink/jars\-f/data/flink-cdc-jobs/sql/cdc_to_kafka.sql

SQL Client embedded 模式会启动一个独立的 MiniCluster 进程来运行作业，既可独立调试，也不会占用已有 Standalone 集群的 Slot 资源。如需与已有集群共享资源，改为提交到远程 SQL Gateway 或使用flink run模式（超出本文范围）。

四、连通性测试：数据打印到控制台

在生产环境正式写入 Kafka 之前，建议先通过 Flink SQL 的print连接器将 MongoDB CDC 数据直接打印到控制台，验证 MongoDB 连接、认证以及数据结构是否正常。

4.1 测试 SQL 脚本

创建文件/data/flink-cdc-jobs/sql/test_print.sql：

-- ============================================================-- 连通性测试：MongoDB CDC 数据 → 控制台打印-- 用途：验证 MongoDB 连接和数据格式，确认无误后再写入 Kafka-- ============================================================SET'parallelism.default'='1';-- MongoDB Source-- 字段顺序与 MongoDB 文档结构保持一致-- gmtCreate、gmtModified 在 MongoDB 中为 ISODate，CDC 读取为 TIMESTAMP(3)-- labels、officialSettings 为复杂对象，序列化为 JSON 字符串CREATETABLEmongo_source(_id STRING,_class STRING,source STRING,creatorId STRING,contentTypeINT,title STRING,description STRING,coverImage STRING,coverMediumUrl STRING,coverSmallUrl STRING,isDefaultItemBOOLEAN,recordStatusINT,coverUpdateMode STRING,isPopularBOOLEAN,itemCountINT,sortOrderINT,labels STRING,hasColumnBOOLEAN,isFeaturedBOOLEAN,contentLevel STRING,isPublicBOOLEAN,searchableBOOLEAN,officialSettings STRING,contentFlag STRING,recommendRule STRING,gmtCreateTIMESTAMP(3),gmtModifiedTIMESTAMP(3),PRIMARYKEY(_id)NOTENFORCED)WITH('connector'='mongodb-cdc','hosts'='<mongo-node-1>:27017,<mongo-node-2>:27017','username'='flink_user','password'='<your_password>','database'='your_database','collection'='your_collection','connection.options'='authSource=admin','scan.startup.mode'='initial','scan.incremental.snapshot.enabled'='true','scan.incremental.snapshot.chunk.size.mb'='64');-- Print Sink：将数据输出到控制台（字段顺序与 MongoDB Source 一致）CREATETABLEprint_sink(_id STRING,_class STRING,source STRING,creatorId STRING,contentTypeINT,title STRING,description STRING,coverImage STRING,coverMediumUrl STRING,coverSmallUrl STRING,isDefaultItemBOOLEAN,recordStatusINT,coverUpdateMode STRING,isPopularBOOLEAN,itemCountINT,sortOrderINT,labels STRING,hasColumnBOOLEAN,isFeaturedBOOLEAN,contentLevel STRING,isPublicBOOLEAN,searchableBOOLEAN,officialSettings STRING,contentFlag STRING,recommendRule STRING,gmtCreateTIMESTAMP(3),gmtModifiedTIMESTAMP(3))WITH('connector'='print','print-identifier'='MongoDB-Collection');-- 输出到控制台（gmtCreate、gmtModified 格式化为 yyyy-MM-dd HH:mm:ss）INSERTINTOprint_sinkSELECT_id,_class,source,creatorId,contentType,title,description,coverImage,coverMediumUrl,coverSmallUrl,isDefaultItem,recordStatus,coverUpdateMode,isPopular,itemCount,sortOrder,labels,hasColumn,isFeatured,contentLevel,