将MongoDB的数据同步至Kafka_数据传输服务(DTS)-阿里云帮助中心

数据传输服务DTS（Data Transmission Service）支持将MongoDB的数据同步到Kafka集群中。本文以云数据库MongoDB版（副本集架构）实例为源库，且以阿里云消息队列Kafka版实例为目标库，为您介绍同步操作的步骤。

前提条件

已创建目标阿里云消息队列Kafka版实例。
说明
源库和目标库支持的版本，请参见同步方案概览。
已在目标阿里云消息队列Kafka版实例中创建用于接收数据的Topic。
若源库为分片集群架构的云数据库MongoDB版，则还需为所有Shard节点申请连接地址，且各Shard的账号和密码需保持一致。申请方法，请参见申请Shard或ConfigServer节点连接地址。

注意事项

类型	说明
源库限制	带宽要求：源库所属的服务器需具备足够的出口带宽，否则将影响数据同步速率。若需要进行编辑（如集合的名称映射），则单次同步任务最多支持同步1000个集合。当超出数量限制，任务提交后会显示请求报错，此时建议您拆分待同步的集合，分批配置多个任务，或者配置整库的同步任务。若源MongoDB为分片集群架构的实例，则待同步集合中的_id字段需具有唯一性，否则可能会导致数据不一致。若源MongoDB为分片集群架构的实例，则源Mongos节点的数量不能超过10个。同时，请确保分片集群架构的MongoDB实例中没有孤立文档，否则可能会导致数据不一致甚至任务失败。更多信息，请参见孤立文档和如何清理MongoDB（分片集群架构）的孤立文档。源库不支持单节点架构的MongoDB、Azure Cosmos DB for MongoDB和弹性集群的Amazon DocumentDB。源库需开启Oplog日志，并确保Oplog日志至少保留7天以上；或者开启变更流（Change Streams），并确保DTS能够通过Change Streams订阅到源库最近7天内的数据变更。否则可能会因无法获取源库的数据变更而导致任务失败，极端情况下甚至可能会导致数据不一致或丢失。由此导致的问题，不在DTS的SLA保障范围内。重要建议通过Oplog日志获取源库的数据变更。仅4.0及以上版本的MongoDB支持通过Change Streams获取数据变更，使用Change Streams获取源库的数据变更不支持双向同步。源库为Amazon DocumentDB（非弹性集群）时，需要手动开启Change Streams，并在配置任务时将迁移方式选择为ChangeStream，将架构类型选择为分片集群架构。源库的操作限制：在全量同步阶段，请勿执行库或集合的结构变更（包含数组类型数据的更新），否则会导致数据同步任务失败或源库与目标库的数据不一致。若仅执行全量数据同步，请勿向源实例中写入新的数据，否则会导致源库与目标库的数据不一致。为实时保持数据一致性，建议选择全量数据同步和增量数据同步。若源MongoDB为分片集群实例，在同步实例运行期间，请勿在源库对待同步的对象执行更改数据分布的相关命令（例如shardCollection、reshardCollection、unshardCollection、moveCollection、movePrimary等），否则可能会导致数据不一致。当源库为分片集群架构的MongoDB时，若源库的均衡器Balancer存在均衡数据的行为，则可能会导致实例产生延迟。不支持通过SRV地址连接MongoDB数据库。
其他限制	仅支持集合级别的同步。不支持同步admin和local库中的数据。若待同步的单条数据超过10 MB，会导致任务失败。若源库为分片集群架构的MongoDB：全量同步期间必须关闭源MongoDB数据库的均衡器（Balancer），直至每个子任务都运行到增量阶段，否则可能会造成数据不一致。关于均衡器的操作，请参见管理MongoDB均衡器Balancer。若增量数据同步的方式为Oplog，则DTS无法保证源库不同分片数据在目标Kafka的写入顺序。不保留事务信息，即源库中的事务同步到目标库时会转变为单条的记录。 DTS实例运行期间，若目标Kafka发生了扩容或缩容（如增加或减少Kafka的Broker节点），需要重启DTS实例。若阿里云消息队列Kafka版实例的Broker数量大于3个，需以专线/VPN网关/智能网关的方式接入DTS。请确保DTS能够正常连接源端和目标端。例如，数据库实例的安全设置、自建Kafka配置文件server.properties中的`listeners`和`advertised.listeners`参数，均未对DTS的访问进行限制。在全量任务运行时，DTS将会占用源和目标库一定的读写资源，可能会导致数据库的负载上升。因此建议您在DTS实例运行前评估源库和目标库的性能，并在业务低峰期运行实例（例如源库和目标库的CPU负载在30%以下）。 DTS将尝试恢复在七天内运行失败的实例。因此，在业务切换至目标实例之前，请务必结束或释放该实例，以避免该实例被自动恢复后导致目标数据库的数据被覆盖。由于DTS增量同步的延迟时间是根据同步到目标库最后一条数据的时间戳和当前时间戳对比得出，源库长时间未执行更新操作可能导致延迟信息不准确。如果任务显示的延迟时间过大，您可以在源库执行一个更新操作来更新延迟信息。若实例运行失败，DTS技术支持人员将在8小时内尝试恢复该实例。在恢复失败实例的过程中，可能会对该实例进行重启、调整参数等操作。说明在调整参数时，仅会修改实例的参数，不会对数据库中的参数进行修改。可能修改的参数，包括但不限于修改实例参数中的参数。

费用说明

同步类型	链路配置费用
全量数据同步	不收费。
增量数据同步	收费，详情请参见计费概述。

同步类型说明

同步类型

说明

全量同步

将源云数据库MongoDB版同步对象的存量数据全部同步到目标Kafka实例中。

说明

支持全量同步DATABASE和COLLECTION。

增量同步

在全量同步的基础上，将源云数据库MongoDB版的增量更新同步到目标Kafka实例中。

使用Oplog

增量同步不支持在任务开始运行后新建的数据库，支持同步的增量更新如下：

CREATE COLLECTION、INDEX
DROP DATABASE、COLLECTION、INDEX
RENAME COLLECTION
在集合中插入、更新、删除文档的操作。

使用ChangeStream

支持同步的增量更新如下：

DROP DATABASE、COLLECTION
RENAME COLLECTION
在集合中插入、更新、删除文档的操作。

数据库账号的权限要求

数据库	所需权限	账号创建及授权方式
源云数据库MongoDB版	待同步库、admin库和local库的read权限。	账号管理

操作步骤

进入目标地域的同步任务列表页面（二选一）。
通过DTS控制台进入
1. 登录数据传输服务DTS控制台。
2. 在左侧导航栏，单击数据同步。
3. 在页面左上角，选择同步实例所属地域。
通过DMS控制台进入
说明
实际操作可能会因DMS的模式和布局不同，而有所差异。更多信息，请参见极简模式控制台和自定义DMS界面布局与样式。
1. 登录DMS数据管理服务。
2. 在顶部菜单栏中，选择Data + AI > 数据传输（DTS） > 数据同步。
3. 在同步任务右侧，选择同步实例所属地域。
单击创建任务，进入任务配置页面。

配置源库及目标库信息。

类别	配置	说明
无	任务名称	DTS会自动生成一个任务名称，建议配置具有业务意义的名称（无唯一性要求），便于后续识别。
源库信息	选择已有连接信息	若您需要使用已录入系统（新建或保存）的数据库实例，请在下拉列表中选择所需的数据库实例，下方的数据库信息将自动进行配置。说明 DMS控制台的配置项为选择DMS数据库实例。若您未将数据库实例录入到系统，或无需使用已录入系统的数据库实例，则需要手动配置下方的数据库信息。
	数据库类型	选择MongoDB。
	接入方式	选择云实例。
	实例地区	选择源云数据库MongoDB版所属的地域。
	是否跨阿里云账号	本示例使用当前阿里云账号下的数据库实例，需选择不跨账号。
	架构类型	本示例选择副本集架构。说明若您的源云数据库MongoDB版为分片集群架构，您还需要填写Shard账号和Shard密码。
	迁移方式	请根据实际情况，选择增量数据同步的方式。 Oplog（推荐）：若源库已开启Oplog日志，则支持此选项。说明本地自建MongoDB和云数据库MongoDB版默认已开启Oplog日志，且使用此方式同步增量数据时增量同步任务的延迟较小（拉取日志的速度较快），因此推荐选择Oplog。 ChangeStream：若源库已开启变更流（Change Streams），则支持此选项。说明源库为Amazon DocumentDB（非弹性集群）时，仅支持选择ChangeStream。源库架构类型选择为分片集群架构，无需填写Shard账号和Shard密码。
	实例ID	选择源云数据库MongoDB版的实例ID。
	鉴权数据库名称	填入源云数据库MongoDB版实例中数据库账号所属的数据库名称，若未修改过则为默认的admin。
	数据库账号	填入源云数据库MongoDB版的数据库账号。权限要求，请参见数据库账号的权限要求。
	数据库密码	填入该数据库账号对应的密码。
	连接方式	DTS支持非加密连接、SSL安全连接和Mongo Atlas SSL三种连接方式。连接方式的选项与接入方式和架构类型有关，请以控制台为准。说明架构类型为分片集群架构，且迁移方式为Oplog的MongoDB数据库，不支持SSL安全连接。若源库为自建（接入方式不为云实例）副本集架构的MongoDB数据库，并且选择了SSL安全连接，DTS还支持上传CA证书对连接进行校验。
目标库信息	选择已有连接信息	若您需要使用已录入系统（新建或保存）的数据库实例，请在下拉列表中选择所需的数据库实例，下方的数据库信息将自动进行配置。说明 DMS控制台的配置项为选择DMS数据库实例。若您未将数据库实例录入到系统，或无需使用已录入系统的数据库实例，则需要手动配置下方的数据库信息。
	数据库类型	选择Kafka。
	接入方式	选择云实例。
	实例地区	选择目标Kafka实例所属的地域。
	Kafka实例ID	选择目标Kafka实例的ID。
	连接方式	请根据业务及安全需求，选择非加密连接或SCRAM-SHA-256。
	Topic	在下拉框中选择用于接收数据的Topic。
	存储DDL的Topic	在下拉框中选择用于存储DDL信息的Topic。说明若未选择，DDL信息默认存储在Topic选择的Topic中。
	是否使用Kafka Schema Registry	Kafka Schema Registry是元数据提供服务层，提供了一个RESTful接口，用于存储和检索Avro Schema。否：不使用Kafka Schema Registry。是：使用Kafka Schema Registry。您需要输入Avro Schema在Kafka Schema Registry注册的URL或IP。

配置完成后，在页面下方单击测试连接以进行下一步。
说明
- 请确保DTS服务的IP地址段能够被自动或手动添加至源库和目标库的安全设置中，以允许DTS服务器的访问。更多信息，请参见添加DTS服务器的IP地址段。
- 若源库或目标库为自建数据库（接入方式不是云实例），则还需要在弹出的DTS服务器访问授权对话框单击测试连接。

配置任务对象。

在对象配置页面，配置待同步的对象。

配置	说明
同步类型	固定选中增量同步。仅支持选中全量同步，不支持库表结构同步。预检查完成后，DTS会将源实例中待同步对象的数据在目标集群中初始化，作为后续增量同步数据的基线数据。
目标已存在表的处理模式	预检查并报错拦截：检查目标数据库中是否有同名的集合。如果目标数据库中没有同名的集合，则通过该检查项目；如果目标数据库中有同名的集合，则在预检查阶段提示错误，数据同步任务不会被启动。说明如果目标库中同名的集合不方便删除或重命名，您可以更改该集合在目标库中的名称，详情请参见设置同步对象在目标实例中的名称。忽略报错并继续执行：跳过目标数据库中是否有同名集合的检查项。警告选择为忽略报错并继续执行，可能导致数据不一致，给业务带来风险，例如：在目标库遇到与源库主键或唯一键的值相同的记录，则会保留目标库中的该条记录，即源库中的该条记录不会同步至目标库中。可能会导致无法初始化数据、只能同步部分的数据或同步失败。
投递到Kafka的数据格式	仅支持选择Canal JSON。说明 Kafka接收到的数据可以分为三种场景。
Kafka压缩格式	根据需求选择Kafka压缩消息的压缩格式。 LZ4（默认）：压缩率较低，压缩速率较高。 GZIP：压缩率较高，压缩速率较低。说明对CPU的消耗较高。 Snappy：压缩率中等，压缩速率中等。
投递到Kafka Partition策略	根据业务需求选择策略。
消息确认机制	根据业务需求选择消息确认机制。
目标库对象名称大小写策略	您可以配置目标实例中同步对象的库名和集合名的英文大小写策略。默认情况下选择DTS默认策略，您也可以选择与源库或目标库默认策略保持一致。更多信息，请参见目标库对象名称大小写策略。
源库对象	在源库对象框中单击待同步对象，然后单击将其移动至已选择对象框。说明同步对象的选择粒度为集合。
已选择对象	本示例无需额外配置。您可以使用映射功能，设置源库中的集合在目标Kafka实例中的映射信息。

单击下一步高级配置，进行高级参数配置。

配置	说明
选择调度该任务的专属集群	DTS默认将任务调度到共享集群上，您无需选择。若您希望任务更加稳定，可以购买专属集群来运行DTS同步任务。更多信息，请参见什么是DTS专属集群。
源库、目标库无法连接后的重试时间	在同步任务启动后，若源库或目标库连接失败则DTS会报错，并会立即进行持续的重试连接，默认持续重试时间为720分钟，您也可以在取值范围（10~1440分钟）内自定义重试时间，建议设置30分钟以上。如果DTS在设置的重试时间内重新连接上源库、目标库，同步任务将自动恢复。否则，同步任务将会失败。说明针对同源或者同目标的多个DTS实例，如DTS实例A和DTS实例B，设置网络重试时间时A设置30分钟，B设置60分钟，则重试时间以低的30分钟为准。由于连接重试期间，DTS将收取任务运行费用，建议您根据业务需要自定义重试时间，或者在源和目标库实例释放后尽快释放DTS实例。
源库、目标库出现其他问题后的重试时间	在同步任务启动后，若源库或目标库出现非连接性的其他问题（如DDL或DML执行异常），则DTS会报错并会立即进行持续的重试操作，默认持续重试时间为10分钟，您也可以在取值范围（1~1440分钟）内自定义重试时间，建议设置10分钟以上。如果DTS在设置的重试时间内相关操作执行成功，同步任务将自动恢复。否则，同步任务将会失败。重要源库、目标库出现其他问题后的重试时间的值需要小于源库、目标库无法连接后的重试时间的值。
是否获取更新操作后的完整文档	增量数据同步阶段，是否将更新操作后对应文档（Document）的完整数据同步到目标端。说明仅当迁移方式选择ChangeStream时，才有此配置项。是：同步更新字段对应文档的完整数据。否：只同步更新字段的数据。
是否限制全量同步速率	在全量同步阶段，DTS将占用源库和目标库一定的读写资源，可能会导致数据库的负载上升。您可以根据实际情况，选择是否对全量同步任务进行限速设置（设置每秒查询源库的速率QPS、每秒全量迁移的行数RPS和每秒全量迁移的数据量(MB)BPS），以缓解目标库的压力。说明仅当同步类型选择了全量同步，才有此配置项。
待同步的数据中，同一张表内主键_id的数据类型是否唯一	待同步的数据中，同一个集合内主键`_id`的数据类型是否唯一。重要请根据实际情况选择，否则可能会导致数据丢失。仅当同步类型选择了全量同步，才有此配置项。是：唯一。在全量同步阶段，DTS将不会扫描源库待同步数据中主键的数据类型；在单个集合中，DTS仅会同步一种数据类型的主键所对应的数据。否：不唯一。在全量同步阶段，DTS将扫描源库待同步数据中主键的数据类型，并同步其所有数据。
是否限制增量同步速率	您也可以根据实际情况，选择是否对增量同步任务进行限速设置（设置每秒增量同步的行数RPS和每秒增量同步的数据量(MB)BPS），以缓解目标库的压力。
环境标签	您可以根据实际情况，选择用于标识实例的环境标签。本示例无需选择。
配置ETL功能	选择是否配置ETL功能。关于ETL的更多信息，请参见什么是ETL。是：配置ETL功能，并在文本框中填写数据处理语句，详情请参见在DTS迁移或同步任务中配置ETL。否：不配置ETL功能。
监控告警	是否设置告警，当同步失败或延迟超过阈值后，将通知告警联系人。不设置：不设置告警。设置：设置告警，您还需要设置告警阈值和告警联系人。更多信息，请参见在配置任务过程中配置监控告警。

保存任务并进行预检查。
- 若您需要查看调用API接口配置该实例时的参数信息，请将鼠标光标移动至下一步保存任务并预检查按钮上，然后单击气泡中的预览OpenAPI参数。
- 若您无需查看或已完成查看API参数，请单击页面下方的下一步保存任务并预检查。
说明
- 在同步作业正式启动之前，会先进行预检查。只有预检查通过后，才能成功启动同步作业。
- 如果预检查失败，请单击失败检查项后的查看详情，并根据提示修复后重新进行预检查。
- 如果预检查产生警告：
  对于不可以忽略的检查项，请单击失败检查项后的查看详情，并根据提示修复后重新进行预检查。
  对于可以忽略无需修复的检查项，您可以依次单击点击确认告警详情、确认屏蔽、确定、重新进行预检查，跳过告警检查项重新进行预检查。如果选择屏蔽告警检查项，可能会导致数据不一致等问题，给业务带来风险。

购买实例。

预检查通过率显示为100%时，单击下一步购买。

在购买页面，选择数据同步实例的计费方式、链路规格，详细说明请参见下表。

类别	参数	说明
信息配置	计费方式	预付费（包年包月）：在新建实例时支付费用。适合长期需求，价格比按量付费更实惠，且购买时长越长，折扣越多。后付费（按量付费）：按小时扣费。适合短期需求，用完可立即释放实例，节省费用。
	资源组配置	实例所属的资源组，默认为default resource group。更多信息，请参见什么是资源管理。
	链路规格	DTS为您提供了不同性能的同步规格，同步链路规格的不同会影响同步速率，您可以根据业务场景进行选择。更多信息，请参见数据同步链路规格说明。
	订购时长	在预付费模式下，选择包年包月实例的时长和数量，包月可选择1~9个月，包年可选择1年、2年、3年和5年。说明该选项仅在付费类型为预付费时出现。

配置完成后，阅读并勾选《数据传输（按量付费）服务条款》。
单击购买并启动，并在弹出的确认对话框，单击确定。
您可在数据同步界面查看具体任务进度。

映射信息

在已选择对象区域框中，将鼠标指针放置在目标Topic名（集合级别）上。
单击目标Topic名后出现的编辑。

在弹出的编辑表对话框中，配置映射信息。

配置	说明
表名称	源集合同步到的目标Topic名称，默认为源库及目标库配置阶段在目标库信息选择的Topic。重要填写的Topic名称必须在目标Kafka实例中真实存在，否则将会导致数据同步失败。若您修改了表名称，数据将会被写入到您填写的Topic中。
过滤条件	详情请参见设置过滤条件。
设置新建Topic的Partition数量	数据写入到目标Topic时的分区数。

单击确定。

投递数据场景

场景一：使用Oplog方式同步增量数据

实例主要配置

迁移方式选择为Oplog。

数据投递示例

源库增量变更类型	源库增量变更语句	目标Topic接收到的数据
`insert`	`db.kafka_test.insert({"cid":"a","person":{"name":"testName","age":NumberInt(18),"skills":["database","ai"]}})`	查看数据（单击展开） `{ "data": [{ "person": { "skills": ["database", "ai"], "name": "testName", "age": 18 }, "_id": { "$oid": "67d27da49591697476e1**" }, "cid": "a" }], "database": "kafkadb", "es": 1741847972000, "gtid": null, "id": 174184797200000**, "isDdl": false, "mysqlType": null, "old": null, "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741847973438, "type": "INSERT" }`
`update $set`	`db.kafka_test.update({"cid":"a"},{$set:{"person.age":NumberInt(20)}})`	查看数据（单击展开） `{ "data": [{ "$set": { "person.age": 20 } }], "database": "kafkadb", "es": 1741848051000, "gtid": null, "id": 174184805100000**, "isDdl": false, "mysqlType": null, "old": [{ "_id": { "$oid": "67d27da49591697476e1**" } }], "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741848051984, "type": "UPDATE" }`
`update $set new filed`	`db.kafka_test.update({"cid":"a"},{$set:{"salary":100}})`	查看数据（单击展开） `{ "data": [{ "$set": { "salary": 100.0 } }], "database": "kafkadb", "es": 1741848146000, "gtid": null, "id": 174184814600000**, "isDdl": false, "mysqlType": null, "old": [{ "_id": { "$oid": "67d27da49591697476e1**" } }], "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741848147734, "type": "UPDATE" }`
`update $unset remove field`	`db.kafka_test.update({"cid":"a"},{$unset:{"salary":1}})`	查看数据（单击展开） `{ "data": [{ "$unset": { "salary": true } }], "database": "kafkadb", "es": 1741848207000, "gtid": null, "id": 174184820700000**, "isDdl": false, "mysqlType": null, "old": [{ "_id": { "$oid": "67d27da49591697476e1**" } }], "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741848208186, "type": "UPDATE" }`
`delete`	`db.kafka_test.deleteOne({"cid":"a"})`	查看数据（单击展开） `{ "data": [{ "_id": { "$oid": "67d27da49591697476e1**" } }], "database": "kafkadb", "es": 1741848289000, "gtid": null, "id": 174184828900000**, "isDdl": false, "mysqlType": null, "old": null, "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741848289798, "type": "DELETE" }`
`ddl drop`	`db.kafka_test.drop()`	查看数据（单击展开） `{ "data": null, "database": "kafkadb", "es": 1741847893000, "gtid": null, "id": 1741847893000000005, "isDdl": true, "mysqlType": null, "old": null, "pkNames": null, "serverId": null, "sql": { "drop": "kafka_test" }, "sqlType": null, "table": "kafka_test", "ts": 1741847893760, "type": "DDL" }`

场景二：使用ChangeStream方式同步增量数据（同步更新字段的数据）

实例主要配置

迁移方式选择为ChangeStream，且是否获取更新操作后的完整文档选择为否。

数据投递示例

源库增量变更类型	源库增量变更语句	目标Topic接收到的数据
`insert`	`db.kafka_test.insert({"cid":"a","person":{"name":"testName","age":NumberInt(18),"skills":["database","ai"]}})`	查看数据（单击展开） `{ "data": [{ "person": { "skills": ["database", "ai"], "name": "testName", "age": 18 }, "_id": { "$oid": "67d27da49591697476e1**" }, "cid": "a" }], "database": "kafkadb", "es": 1741847972000, "gtid": null, "id": 174184797200000**, "isDdl": false, "mysqlType": null, "old": null, "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741847973803, "type": "INSERT" }`
`update $set`	`db.kafka_test.update({"cid":"a"},{$set:{"person.age":NumberInt(20)}})`	查看数据（单击展开） `{ "data": [{ "$set": { "person.age": 20 } }], "database": "kafkadb", "es": 1741848051000, "gtid": null, "id": 174184805100000**, "isDdl": false, "mysqlType": null, "old": [{ "_id": { "$oid": "67d27da49591697476e1**" } }], "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741848052912, "type": "UPDATE" }`
`update $set new filed`	`db.kafka_test.update({"cid":"a"},{$set:{"salary":100}})`	查看数据（单击展开） `{ "data": [{ "$set": { "salary": 100.0 } }], "database": "kafkadb", "es": 1741848146000, "gtid": null, "id": 174184814600000**, "isDdl": false, "mysqlType": null, "old": [{ "_id": { "$oid": "67d27da49591697476e1**" } }], "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741848148056, "type": "UPDATE" }`
`update $unset remove field`	`db.kafka_test.update({"cid":"a"},{$unset:{"salary":1}})`	查看数据（单击展开） `{ "data": [{ "$unset": { "salary": 1 } }], "database": "kafkadb", "es": 1741848207000, "gtid": null, "id": 174184820700000**, "isDdl": false, "mysqlType": null, "old": [{ "_id": { "$oid": "67d27da49591697476e1**" } }], "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741848209142, "type": "UPDATE" }`
`delete`	`db.kafka_test.deleteOne({"cid":"a"})`	查看数据（单击展开） `{ "data": [{ "_id": { "$oid": "67d27da49591697476e1**" } }], "database": "kafkadb", "es": 1741848289000, "gtid": null, "id": 174184828900000**, "isDdl": false, "mysqlType": null, "old": null, "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741848290254, "type": "DELETE" }`
`ddl drop`	`db.kafka_test.drop()`	查看数据（单击展开） `{ "data": null, "database": "kafkadb", "es": 1741847893000, "gtid": null, "id": 174184789300000****, "isDdl": true, "mysqlType": null, "old": null, "pkNames": null, "serverId": null, "sql": { "drop": "kafka_test" }, "sqlType": null, "table": "kafka_test", "ts": 1741847894679, "type": "DDL" }`

场景三：使用ChangeStream方式同步增量数据（同步更新字段对应文档的完整数据）

实例主要配置

迁移方式选择为ChangeStream，且是否获取更新操作后的完整文档选择为是。

数据投递示例

源库增量变更类型	源库增量变更语句	目标Topic接收到的数据
`insert`	`db.kafka_test.insert({"cid":"a","person":{"name":"testName","age":NumberInt(18),"skills":["database","ai"]}})`	查看数据（单击展开） `{ "data": [{ "person": { "skills": ["database", "ai"], "name": "testName", "age": 18 }, "_id": { "$oid": "67d27da49591697476e1**" }, "cid": "a" }], "database": "kafkadb", "es": 1741847972000, "gtid": null, "id": 174184797200000**, "isDdl": false, "mysqlType": null, "old": null, "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741847973128, "type": "INSERT" }`
`update $set`	`db.kafka_test.update({"cid":"a"},{$set:{"person.age":NumberInt(20)}})`	查看数据（单击展开） `{ "data": [{ "person": { "skills": ["database", "ai"], "name": "testName", "age": 20 }, "_id": { "$oid": "67d27da49591697476e1**" }, "cid": "a" }], "database": "kafkadb", "es": 1741848051000, "gtid": null, "id": 174184805100000, "isDdl": false, "mysqlType": null, "old": [{ "_id": { "$oid": "67d27da49591697476e1**" } }], "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741848052219, "type": "UPDATE" }`
`update $set new filed`	`db.kafka_test.update({"cid":"a"},{$set:{"salary":100}})`	查看数据（单击展开） `{ "data": [{ "person": { "skills": ["database", "ai"], "name": "testName", "age": 20 }, "_id": { "$oid": "67d27da49591697476e1**" }, "salary": 100.0, "cid": "a" }], "database": "kafkadb", "es": 1741848146000, "gtid": null, "id": 174184814600000, "isDdl": false, "mysqlType": null, "old": [{ "_id": { "$oid": "67d27da49591697476e1**" } }], "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741848147327, "type": "UPDATE" }`
`update $unset remove field`	`db.kafka_test.update({"cid":"a"},{$unset:{"salary":1}})`	查看数据（单击展开） `{ "data": [{ "person": { "skills": ["database", "ai"], "name": "testName", "age": 20 }, "_id": { "$oid": "67d27da49591697476e1**" }, "cid": "a" }], "database": "kafkadb", "es": 1741848207000, "gtid": null, "id": 174184820700000, "isDdl": false, "mysqlType": null, "old": [{ "_id": { "$oid": "67d27da49591697476e1**" } }], "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741848208401, "type": "UPDATE" }`
`delete`	`db.kafka_test.deleteOne({"cid":"a"})`	查看数据（单击展开） `{ "data": [{ "_id": { "$oid": "67d27da49591697476e1**" } }], "database": "kafkadb", "es": 1741848289000, "gtid": null, "id": 174184828900000**, "isDdl": false, "mysqlType": null, "old": null, "pkNames": ["_id"], "serverId": null, "sql": null, "sqlType": null, "table": "kafka_test", "ts": 1741848290499, "type": "DELETE" }`
`ddl drop`	`db.kafka_test.drop()`	查看数据（单击展开） `{ "data": null, "database": "kafkadb", "es": 1741847893000, "gtid": null, "id": 174184789300000****, "isDdl": true, "mysqlType": null, "old": null, "pkNames": null, "serverId": null, "sql": { "drop": "kafka_test" }, "sqlType": null, "table": "kafka_test", "ts": 1741847894045, "type": "DDL" }`

特殊情况

注意事项

当更新事件的 fullDocument 字段缺失时，数据的投递结果等同于使用Oplog方式同步增量数据。

示例

源库基础数据

源库增量变更语句

目标Topic接收到的数据

use admin
db.runCommand({ enablesharding:"dts_test" })

use dts_test
sh.shardCollection("dts_test.cstest",{"name":"hashed"})
db.cstest.insert({"_id":1,"name":"a"})
db.cstest.updateOne({"_id":1,"name":"a"},{$set:{"name":"b"}})

查看数据（单击展开）

{
	"data": [{
		"$set": {
			"name": "b"
		}
	}],
	"database": "dts_test",
	"es": 1740720994000,
	"gtid": null,
	"id": 174072099400000****,
	"isDdl": false,
	"mysqlType": null,
	"old": [{
		"name": "a",
		"_id": 1.0
	}],
	"pkNames": ["_id"],
	"serverId": null,
	"sql": null,
	"sqlType": null,
	"table": "cstest",
	"ts": 1740721007099,
	"type": "UPDATE"
}

常见问题

是否支持修改Kafka压缩格式？
支持，您可以使用修改同步对象功能进行修改。
是否支持修改消息确认机制？
支持，您可以使用修改同步对象功能进行修改。

前提条件

注意事项

费用说明

同步类型说明

使用Oplog

使用ChangeStream

数据库账号的权限要求

操作步骤

通过DTS控制台进入

通过DMS控制台进入

映射信息

投递数据场景

场景一：使用Oplog方式同步增量数据

实例主要配置

数据投递示例

场景二：使用ChangeStream方式同步增量数据（同步更新字段的数据）

实例主要配置

数据投递示例

场景三：使用ChangeStream方式同步增量数据（同步更新字段对应文档的完整数据）

实例主要配置

数据投递示例

特殊情况

注意事项

示例

常见问题