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/。核心技术栈组件版本用途Apache Flink1.17.1（Standalone）流式计算引擎flink-sql-connector-mongodb-cdc2.4.0MongoDB CDC Source 连接器flink-sql-connector-kafka1.17.1Kafka Sink 连接器MongoDB≥ 4.0（副本集/分片集）数据源（需开启 Change Stream）Apache Kafka≥ 2.4目标端消息队列数据链路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 前置条件MongoDB 必须为副本集（Replica Set）或分片集群（Sharded Cluster），单机模式不支持 Change Stream。MongoDB 版本不低于 4.0（Change Stream 自 3.6 引入，但 4.0+ 稳定性更高）。Flink 集群所有节点（Master + Worker）都必须能网络连通 MongoDB 各节点。CDC 连接器在全量快照阶段需要 JobManager 直连 MongoDB 获取分片元数据，仅开通 Worker 节点网络会导致 Master 节点MongoTimeoutException超时报错。实际环境连接信息：参数值节点地址mongo-node-1:27017,mongo-node-2:27017数据库your_database集合your_collection用户名flink_user密码your_password验证副本集状态与连通性：# 测试连通性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 前置条件Kafka 集群版本不低于 2.4。创建目标 Topic（4 分区、2 副本）：# 创建 Topickafka-topics.sh --bootstrap-serverkafka-broker-1:9092\--create--topiccdc_sync_topic\--partitions4--replication-factor2# 验证 Topic 已创建kafka-topics.sh --bootstrap-serverkafka-broker-1:9092\--describe--topiccdc_sync_topic# 删除 Topic（如需重建）kafka-topics.sh --bootstrap-serverkafka-broker-1:9092\--delete--topiccdc_sync_topicFlink 集群所在机器能访问 Kafka 各 Broker。实际环境连接信息：参数值Broker 地址kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092,kafka-broker-4:9092Topiccdc_sync_topic验证连通性：# 测试 Broker 连通性telnetkafka-broker-16667telnetkafka-broker-26667telnetkafka-broker-36667telnetkafka-broker-46667# 列出 Topickafka-topics.sh --bootstrap-serverkafka-broker-1:9092--list# 消费验证（--max-messages 限制条数，避免数据量大导致终端卡顿）kafka-console-consumer.sh\--bootstrap-serverkafka-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 包清单JAR 包版本说明flink-sql-connector-kafka1.17.1Kafka Sink 连接器flink-sql-connector-mongodb-cdc2.4.0MongoDB CDC Source 连接器3.2 下载链接(1) flink-sql-connector-kafkaMaven 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-cdcMaven 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-*.jar3.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.sqlSQL 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,