Kafka 到 MongoDB

选择对端数据库：

MongoDB

数据链路

基本功能

功能	说明
Incremental Data Sync	Allow subscribing to messages from the source Topic and transforming them into DML operations, namely, INSERT, UPDATE, DELETE.
Subscription Modification	Add, delete, or modify the subscribed topics. For more information, see Modify Subscription.
Position Resetting	Reset positions by timestamp to consume the data in a past period again.

高级功能

功能	说明
消息格式	支持以下消息格式，文档：消息格式说明 CloudCanal内置格式 AlibabaCanal兼容格式
Pre-Creation of Collection in the Target	Automatic collection creation is not supported when moving data from Kafka/AutoMQ to 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 节点

任务参数

参数名称	说明
totalDataInMemMb	攒批写入，内存中最大数据容量，超过此容量或超过 asyncFlushIntervalSec 则刷出数据到写入队列
asyncFlushIntervalSec	攒批写入，等待刷出的间隔时间，超过此时间或超过 totalDataInMemMb 则刷出数据到写入队列
flushBatchMb	单表最大攒批容量，超过此容量则刷出数据到写入队列
realFlushPauseSec	刷出数据到 MongoDB 的等待时间，0 则不等待
enableTimeZoneProcess	是否对时间字段进行时区转换
timezone	需要转换的目标时区（默认 UTC）
enableBatchApply	启用批量写入模式 （默认关闭）

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