Kafka 到 MongoDB

选择对端数据库：

MongoDB

数据链路

基本功能

功能	说明
增量实时同步	支持订阅源端 Topic 的消息，并转换为 INSERT、UPDATE、DELETE DML
修改订阅	新增、删除、修改订阅 Topic，文档：修改订阅
重置位点	按 时间戳 回溯位点，重新消费过去一段时间的数据

高级功能

功能	说明
消息格式	支持以下消息格式，文档：消息格式说明 CloudCanal内置格式 AlibabaCanal兼容格式
目标端需要提前创建集合	Kafka / AutoMQ 到 MongoDB 不支持自动创建集合

限制和注意点

限制项	说明
目标端需要提前创建表	仅支持消息自动创建 Topic
原始消息格式	仅支持 Kafka 到 Kafka，且两端的消息格式都需要选择 原始消息格式

使用示例

标题	详情
跨互联网数据互通 (Kafka)	文档：跨互联网数据互通 (Kafka)
Kafka 数据中转校验	文档：Kafka 数据中转校验

链路FAQ

• 如何打开 Kafka 源端心跳?

---

源端数据源

前置条件

条件	说明
网络准备	迁移同步节点（sidecar）可连接 Kafka 各节点

任务参数

参数名称	说明
schemaFormat	MQ 消息格式，文档：消息格式说明
consumerGroupId	Kafka 消费组 Id
consumeParallel	消费 Kafka 的并行度
sessonTimeoutMs	Kafka Session 超时时间（毫秒）
maxPollRecords	Kafka 一次最大拉取消息数量
dbHeartbeatIntervalSec	配置对源端数据库发起心跳操作的间隔时长
dbHeartbeatToleranceStep	配置对源端数据库心跳操作可容忍的位点差值
customClientProps	自定义传入到 Kafka Client 参数，JSON 格式，key为参数名，value为参数值。此配置项以最高优先级生效。例如：AWS IAM 访问控制

Tips: 通用参数配置请参考 通用参数及功能

---

目标端数据源

前置条件

条件	说明
账号权限	文档：MongoDB 需要的权限
网络准备	迁移同步节点（sidecar）可连接 MongoDB/DocumentDB 节点
双向同步准备	文档：MongoDB 双向同步准备

任务参数

参数名称	说明
totalDataInMemMb	攒批写入，内存中最大数据容量，超过此容量或超过 asyncFlushIntervalSec 则刷出数据到写入队列
asyncFlushIntervalSec	攒批写入，等待刷出的间隔时间，超过此时间或超过 totalDataInMemMb 则刷出数据到写入队列
flushBatchMb	单表最大攒批容量，超过此容量则刷出数据到写入队列
realFlushPauseSec	刷出数据到 MongoDB 的等待时间，0 则不等待
enableTimeZoneProcess	是否对时间字段进行时区转换
timezone	需要转换的目标时区（默认 UTC）
enableBatchApply	启用批量写入模式 （默认关闭）
deCycle	是否开启双向同步防循环，默认关闭
deCycleCollection	存储防循环事务 marker 的 collection，默认 cloudcanal.__cc_tx_de_cycle
deCycleMarkerTtlDays	双向同步 marker 的保留天数，默认 7 天。应大于任务允许的最大停机、积压和位点回溯时间

Tips: 通用参数配置请参考 通用参数及功能