Synchronize data from MongoDB to Kafka - Data Transmission Service

Data Transmission Service (DTS) permite transmitir datos de cambios desde ApsaraDB for MongoDB hacia una instancia de Message Queue for Apache Kafka. A continuación se describe cómo crear una tarea de sincronización con un conjunto de réplicas de MongoDB como origen y una instancia de Kafka como destino.

Esta tarea abarca lo siguiente:

Arquitecturas compatibles: conjunto de réplicas y clúster fragmentado.
Métodos de sincronización incremental: Oplog (recomendado) y Change Streams.
Formato de datos: Canal JSON enviado a los topics de Kafka.
Alcance de la sincronización: nivel de colección; sincronización completa e incremental.
Facturación: la sincronización completa de datos es gratuita; la sincronización incremental tiene costo.

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Una instancia de Message Queue for Apache Kafka instance.
Un topic creado en la instancia de Kafka de destino para recibir los datos.
(Solo para clústeres fragmentados) Endpoints solicitados para todos los nodos shard. Todos los nodos shard deben compartir la misma contraseña de cuenta y el mismo endpoint. Consulte Apply for an endpoint for a shard.

Para conocer las versiones de bases de datos de origen y destino compatibles, consulte Descripción general de las soluciones de sincronización .

Límites

Revise los siguientes límites antes de crear una tarea de sincronización.

Límites de la base de datos de origen

El servidor de origen debe tener suficiente ancho de banda de salida. Un ancho de banda insuficiente reduce la velocidad de sincronización.
Al configurar la asignación de nombres para las colecciones, una sola tarea puede sincronizar hasta 1,000 colecciones. Si supera este límite, se genera un error de solicitud. Para sincronizar más de 1,000 colecciones, cree varias tareas o sincronice toda la base de datos sin asignación de nombres.
Para orígenes de clúster fragmentado: el campo _id de cada colección sincronizada debe ser único; de lo contrario, pueden producirse inconsistencias en los datos.
Para orígenes de clúster fragmentado: la cantidad de nodos mongos no puede superar los 10 y la instancia no debe contener documentos huérfanos. Consulte FAQ topic para saber cómo eliminar documentos huérfanos.
No se admiten como origen las instancias independientes de ApsaraDB for MongoDB, los clústeres de Azure Cosmos DB for MongoDB ni los clústeres elásticos de Amazon DocumentDB.
DTS no puede conectarse a MongoDB mediante un endpoint SRV.
El oplog debe estar habilitado y conservar los datos de registro durante como mínimo 7 días, o bien deben estar habilitados los change streams para que DTS pueda suscribirse a los cambios de datos de los últimos 7 días. Si no se cumple ninguna de estas condiciones, DTS podría no recuperar los cambios del origen, lo que provocaría pérdida o inconsistencia de datos fuera del alcance del acuerdo de nivel de servicio (SLA).
Importante
- Utilice el oplog para registrar los cambios de la base de datos de origen (recomendado). - Los change streams requieren MongoDB 4.0 o posterior. La sincronización bidireccional no es compatible cuando se usan change streams. - En clústeres de Amazon DocumentDB no elásticos, habilite los change streams, establezca Migration Method en ChangeStream y Architecture en Sharded Cluster.
Durante la sincronización completa de datos, no modifique los esquemas de bases de datos ni de colecciones, y tampoco cambie datos del tipo ARRAY.
En orígenes de clúster fragmentado, no ejecute los siguientes comandos durante la sincronización: shardCollection, reshardCollection, unshardCollection, moveCollection ni movePrimary. Estos comandos pueden causar inconsistencias en los datos.
Si la base de datos de origen es una instancia de MongoDB con arquitectura de clúster fragmentado y su balanceador distribuye datos, la instancia puede presentar latencia.

Otros límites

Solo se admite la sincronización a nivel de colección.
Las bases de datos admin, config y local no se pueden sincronizar.
Un registro individual que supere los 10 MB provoca un error en la tarea.
Las transacciones no se conservan. DTS convierte cada transacción en registros individuales en el destino.
Si se agregan o eliminan nodos broker en la instancia de Kafka de destino mientras se ejecuta una tarea de DTS, reinicie dicha tarea.
Asegúrese de que DTS pueda conectarse a las instancias de origen y destino. Por ejemplo, compruebe que la configuración de seguridad de las instancias de base de datos y los parámetros listeners y advertised.listeners del archivo server.properties de una instancia de Kafka autogestionada no restrinjan el acceso desde DTS.
Ejecute la sincronización en horarios de baja demanda, cuando la utilización de CPU de las bases de datos de origen y destino sea inferior al 30%.
DTS reintenta automáticamente las instancias con errores que llevan menos de 7 days en ejecución. Antes de dirigir el tráfico al destino, detenga o libere la instancia de sincronización para evitar que la reanudación automática sobrescriba los datos de destino.
DTS calcula la latencia de la sincronización incremental de datos con base en la marca de tiempo de los últimos datos sincronizados en la base de datos de destino y la marca de tiempo actual de la base de datos de origen. Si no se ejecutan operaciones de actualización en el origen durante un periodo prolongado, la latencia mostrada puede ser inexacta. Cuando la latencia de la tarea de sincronización sea excesivamente alta, ejecute una operación de actualización en la base de datos de origen para corregirla.
Si una tarea de DTS presenta errores de ejecución, el soporte técnico intentará restaurarla en un plazo de 8 horas. Durante la restauración, la tarea podría reiniciarse y sus parámetros podrían modificarse. Solo se modifican los parámetros de la tarea de DTS, nunca los de las bases de datos. Entre los parámetros sujetos a modificación se incluyen, entre otros, los descritos en la sección «Modificar parámetros de instancia».
En orígenes de clúster fragmentado donde Oplog es el método de sincronización incremental, DTS no garantiza el orden de escritura entre shards hacia el topic de Kafka de destino.
En orígenes de clúster fragmentado, deshabilite el balanceador de MongoDB durante la sincronización completa de datos. Vuelva a habilitarlo únicamente cuando finalice la sincronización completa y comience la incremental. Consulte Manage the ApsaraDB for MongoDB balancer.

Facturación

Synchronization type	Cost
Sincronización completa de datos	Gratuita
Sincronización incremental de datos	Con cargo. Consulte Descripción general de la facturación

Tipos de sincronización y operaciones compatibles

La Full synchronization copia todos los datos históricos de las colecciones de MongoDB seleccionadas al topic de Kafka de destino. DTS admite objetos DATABASE y COLLECTION.

La Incremental synchronization envía continuamente eventos de cambio después de que finaliza la sincronización completa. Las operaciones compatibles dependen del método de sincronización incremental:

Operation	Oplog	Change streams
INSERT	Compatible	Compatible
UPDATE	Compatible	Compatible
DELETE	Compatible	Compatible
CREATE COLLECTION / INDEX	Compatible	No compatible
DROP DATABASE / COLLECTION / INDEX	Compatible	Solo DROP DATABASE y COLLECTION
RENAME COLLECTION	Compatible	Compatible

La sincronización incremental no detecta las bases de datos creadas después de iniciar la tarea.

Permisos obligatorios de la cuenta de base de datos

Database	Required permissions
ApsaraDB for MongoDB de origen	Permiso de lectura en las bases de datos que se van a sincronizar, así como en las bases de datos `admin` y `local`

Para consultar las instrucciones de creación de cuentas, vea Account management.

Nota

Si utiliza ChangeStream como método de sincronización incremental, la cuenta de la base de datos de origen requiere permisos de lectura de Change Streams en toda la instancia (por ejemplo, readAnyDatabase). Si el origen es una instancia de ApsaraDB for MongoDB con una cuenta personalizada, también debe otorgarle permiso de lectura en la base de datos admin. Para más detalles, consulte Permissions of the root account specified during instance creation.

Creación de una tarea de sincronización

Paso 1: Abrir la página Data Synchronization

Utilice la consola de DTS o la consola de DMS.

Consola de DTS

Inicie sesión en la consola de DTS.consola de DTS
En el panel de navegación izquierdo, haga clic en Data Synchronization.
En la esquina superior izquierda, seleccione la región donde reside la tarea de sincronización.

Consola de DMS

Los pasos pueden variar según el modo de la consola de DMS. Consulte Simple mode y Customize the layout and style of the DMS console .

Inicie sesión en la consola de DMS.consola de DMS
En la barra de navegación superior, coloque el puntero sobre Data + AI y elija DTS (DTS) > Data Synchronization.
En la lista desplegable situada a la derecha de Data Synchronization Tasks, seleccione la región.

Paso 2: Configurar las bases de datos de origen y destino

Haga clic en Create Task y configure los parámetros descritos en las siguientes tablas.

General

Parameter	Description
Task Name	Nombre de la tarea de DTS. El sistema genera uno automáticamente. Especifique un nombre descriptivo para identificar la tarea. No es obligatorio que sea único.

Base de datos de origen

Parameter	Description
Select Existing Connection	Si la instancia de origen está registrada en DTS, selecciónela en la lista desplegable. DTS completará automáticamente los campos restantes. De lo contrario, configure los campos siguientes. En la consola de DMS, seleccione la opción correspondiente en la lista Select a DMS database instance.
Database Type	Seleccione MongoDB.
Access Method	Seleccione Alibaba Cloud Instance.
Instance Region	Seleccione la región de la instancia de ApsaraDB for MongoDB de origen.
Replicate Data Across Alibaba Cloud Accounts	Seleccione No si la base de datos de origen pertenece a la cuenta actual de Alibaba Cloud.
Architecture	Seleccione Replica Set para este ejemplo. Si el origen es un Sharded Cluster, introduzca también la Shard account y la Shard password.
Migration Method	Método para sincronizar datos incrementales. Se recomienda Oplog cuando la función oplog está habilitada. ChangeStream está disponible cuando los change streams están habilitados en el origen. Consulte Change Streams para más detalles. Notas: En clústeres de Amazon DocumentDB no elásticos, solo se admite ChangeStream. Cuando Architecture se establece en Sharded Cluster y Migration Method en ChangeStream, los campos Shard account y Shard password no son obligatorios.
Instance ID	Seleccione el identificador de la instancia de ApsaraDB for MongoDB de origen.
Authentication Database	Base de datos que contiene las credenciales de la cuenta. El valor predeterminado es `admin`.
Database Account	Cuenta de la base de datos de origen. Consulte Required database account permissions.
Database Password	Contraseña de la cuenta de la base de datos.
Encryption	Tipo de cifrado de la conexión: Non-encrypted, SSL-encrypted o Mongo Atlas SSL. Las opciones disponibles dependen de los valores de Access Method y Architecture. Notas: La opción SSL-encrypted no está disponible cuando Architecture es Sharded Cluster y Migration Method es Oplog. En instalaciones de MongoDB autogestionadas con arquitectura Replica Set y SSL-encrypted seleccionado, cargue un certificado CA para verificar la conexión.

Base de datos de destino

Parameter	Description
Select Existing Connection	Si la instancia de Kafka de destino está registrada en DTS, selecciónela en la lista desplegable. De lo contrario, configure los campos siguientes.
Database Type	Seleccione Kafka.
Access Method	Seleccione Alibaba Cloud Instance.
Instance Region	Seleccione la región de la instancia de Kafka de destino.
Kafka Instance ID	Seleccione el identificador de la instancia de Kafka de destino.
Encryption	Seleccione Non-encrypted o SCRAM-SHA-256 según sus requisitos de seguridad.
Topic	Seleccione el topic que recibirá los datos sincronizados.
Use Kafka Schema Registry	Kafka Schema Registry es un servicio de metadatos RESTful para almacenar y recuperar esquemas Avro. Seleccione No para omitirlo, o Yesconfiguración de notificaciones de alerta y proporcione la URL o dirección IP de Schema Registry.

Paso 3: Probar la conectividad

Haga clic en Test Connectivity and Proceed en la parte inferior de la página.

DTS agrega automáticamente sus bloques CIDR de servidor a la configuración de seguridad de las bases de datos de origen y destino, si los permisos lo permiten. Para configurarlos manualmente, consulte Add the CIDR blocks of DTS servers .

En bases de datos autogestionadas que no utilizan Alibaba Cloud Instance como método de acceso, haga clic en Test Connectivity dentro del cuadro de diálogo CIDR Blocks of DTS Servers .

Paso 4: Configurar los objetos que se van a sincronizar

En el paso Configure Objects, establezca los siguientes parámetros.

Parámetro	Descripción
Synchronization Types	La opción Incremental Data Synchronization viene seleccionada de forma predeterminada. También puede seleccionar Full Data Synchronization. La opción Schema Synchronization no está disponible. Cuando se habilita la sincronización completa, DTS copia primero todos los datos históricos antes de iniciar la sincronización incremental.
Processing Mode of Conflicting Tables	Precheck and Report Errors: la comprobación previa falla si el destino contiene colecciones con los mismos nombres que el origen. Utilice la asignación de nombres de objetos para resolver conflictos de nombres. Ignore Errors and Proceed: omite la comprobación de conflictos de nombres. Si un registro del destino tiene la misma clave principal o clave única que un registro de origen, se conserva el registro de destino. Aviso Esta opción puede provocar inconsistencias en los datos.
Data Format in Kafka	Solo se admite Canal JSON.
Kafka Data Compression Format	Formato de compresión de los datos escritos en Kafka. Opciones: LZ4 (predeterminado: baja relación de compresión, alta velocidad), GZIP (alta relación de compresión, baja velocidad, alto uso de CPU) y Snappy (equilibrado).
Policy for Shipping Data to Kafka Partitions	Seleccione una directiva de enrutamiento de particiones según sus necesidades.
Message acknowledgement mechanism	Seleccione un message acknowledgment mechanism según sus necesidades.
Topic That Stores DDL Information	Seleccione un topic para almacenar los eventos DDL. Si deja el campo vacío, los eventos DDL se escriben en el mismo topic que los eventos de datos.
Capitalization of Object Names in Destination Instance	Controla el uso de mayúsculas y minúsculas en los nombres de bases de datos y colecciones del destino. El valor predeterminado es DTS default policy. Consulte Specify the capitalization of object names in the destination instance.
Source Objects	Seleccione los objetos en la sección Source Objects y haga clic en el ícono de flecha para moverlos a Selected Objects. Solo se admite la selección a nivel de colección.

Paso 5: Configurar los ajustes avanzados

Haga clic en Next: Advanced Settings y configure los siguientes parámetros.

Parameter	Description
Dedicated Cluster for Task Scheduling	De forma predeterminada, DTS programa la tarea en el clúster compartido. Adquiera un clúster dedicado para obtener mayor estabilidad.
Retry Time for Failed Connections	Tiempo durante el cual DTS reintenta la conexión cuando el origen o el destino no son accesibles. Rango: de 10 a 1440 minutos. Valor predeterminado: 720 minutos. Se recomienda establecer este parámetro en un valor superior a 30. Si especifica tiempos de reintento distintos para varias tareas que comparten la misma base de datos, se aplica el valor más corto. DTS cobra por el tiempo de reintento.
Retry Time for Other Issues	Tiempo durante el cual DTS reintenta cuando fallan operaciones DDL o DML. Rango: de 1 a 1440 minutos. Valor predeterminado: 10 minutos. Establézcalo en como mínimo 10 minutos. Debe ser menor que Retry Time for Failed Connections.
Obtain the entire document after it is updated	Disponible solo cuando Migration Method es ChangeStream. Yes: sincroniza el documento completo en cada actualización, lo que puede aumentar la carga del origen y generar latencia. Si DTS no logra recuperar el documento completo, solo se envían los campos actualizados. No: sincroniza únicamente los campos modificados.
Enable Throttling for Full Data Synchronization	Al habilitar esta opción, configure QPS to the source database, RPS of Full Data Migration y Data migration speed for full migration (MB/s) para reducir la carga en el destino. Disponible solo cuando se selecciona Full Data Synchronization.
Only one data type for primary key _id in a table of the data to be synchronized	Disponible solo cuando se selecciona Full Data Synchronization. Yes: DTS omite el análisis del tipo de dato del campo `_id` durante la sincronización completa. No: DTS analiza el tipo de dato del campo `_id`.
Enable Throttling for Incremental Data Synchronization	Al habilitar esta opción, configure RPS of Incremental Data Synchronization y Data synchronization speed for incremental synchronization (MB/s) para reducir la carga en el destino.
Environment Tag	Etiqueta opcional para identificar la instancia.
Configure ETL	Habilítela para aplicar lógica de extracción, transformación y carga (ETL). Introduzca las instrucciones de procesamiento de datos en el editor de código. Consulte Configure ETL in a data migration or data synchronization task.
Monitoring and Alerting	Al habilitar esta opción, especifique un umbral de alerta y la configuración de notificaciones. DTS avisa a los contactos de alerta cuando la tarea falla o la latencia de sincronización supera el umbral. Consulte Configure monitoring and alerting.

Paso 6: Ejecutar la comprobación previa

Haga clic en Next: Save Task Settings and Precheck.

Para previsualizar los parámetros de API de esta configuración, coloque el puntero sobre el botón y haga clic en Preview OpenAPI parameters antes de continuar.

La tarea no puede iniciarse hasta que la comprobación previa se complete correctamente.

Si la comprobación previa falla, haga clic en View Details junto a cada elemento con errores, resuelva el problema y luego haga clic en Precheck Again .

Si se genera una alerta para algún elemento: corrija las alertas que no se puedan ignorar antes de continuar. Para las alertas que sí se puedan ignorar, haga clic en Confirm Alert Details , luego en Ignore dentro del cuadro de diálogo, confirme y haga clic en Precheck Again . Ignorar alertas puede provocar inconsistencias en los datos.

Paso 7: Adquirir e iniciar la instancia

Espere a que Success Rate alcance el 100% y luego haga clic en Next: Purchase Instance.

En la página buy, configure los siguientes parámetros.

Parameter	Description
Billing Method	Subscription: pago por adelantado; más rentable para uso a largo plazo. Pay-as-you-go: facturación por hora; adecuado para uso a corto plazo. Libere la instancia cuando ya no la necesite para detener la facturación.
Resource Group Settings	Resource group de la instancia de sincronización. Valor predeterminado: default resource group. Consulte What is Resource Management?.
Instance Class	Nivel de velocidad de sincronización. Consulte Instance classes of data synchronization instances.
Subscription Duration	Disponible para la facturación por Subscription. Opciones: de 1 a 9 meses, 1 año, 2 años, 3 años o 5 años.

Lea y acepte los Data Transmission Service (Pay-as-you-go) Service Terms.
Haga clic en Buy and Start y luego en OK en el cuadro de diálogo de confirmación.

La tarea aparecerá en la lista de tareas. Supervise su avance desde ahí.

Configuración de la asignación de colecciones a topics

De forma predeterminada, todas las colecciones se escriben en el topic seleccionado durante la configuración de la base de datos de destino. Para dirigir una colección específica a otro topic, siga estos pasos:

En el área Selected Objects, coloque el puntero sobre el nombre del topic de destino a nivel de colección.
Haga clic en Edit junto al nombre del topic.

En el cuadro de diálogo Edit Table, configure los siguientes ajustes.

Parameter	Description
Name of target Topic	Topic de destino para esta colección. El topic debe existir en la instancia de Kafka. Si lo modifica, los datos se escribirán en el nuevo topic. Valor predeterminado: el topic seleccionado durante la configuración de la base de datos de destino.
Filter Conditions	Filtro de filas opcional. Consulte Filter task data with an SQL condition.
Number of Partitions	Número de particiones para escribir datos en el topic.

Haga clic en OK.

Ejemplos de entrega de datos

Cada cambio incremental de MongoDB se serializa como un mensaje Canal JSON y se entrega al topic de Kafka configurado. La estructura del mensaje varía según el método de sincronización incremental y los ajustes de actualización.

Elija su escenario

Goal	Configuration
Sincronización de baja latencia con compatibilidad total para DDL	Establezca Migration Method en Oplog (escenario 1)
Change Streams con actualizaciones parciales de documentos	Establezca Migration Method en ChangeStream y Obtain the entire document after it is updated en No (escenario 2)
Change Streams con el documento completo en cada actualización	Establezca Migration Method en ChangeStream y Obtain the entire document after it is updated en Yes (escenario 3)

Campos del mensaje Canal JSON

Los tres escenarios utilizan la misma envoltura Canal JSON de nivel superior. A continuación se muestran los campos que aparecen en todos los mensajes.

Field	Type	Description
`database`	string	Nombre de la base de datos de origen
`table`	string	Nombre de la colección de origen
`type`	string	Tipo de operación: `INSERT`, `UPDATE`, `DELETE` o `DDL`
`isDdl`	boolean	`true` para eventos DDL (eliminación o cambio de nombre de una colección); `false` para eventos DML
`es`	number	Marca de tiempo del evento en la base de datos de origen (milisegundos Unix)
`ts`	number	Marca de tiempo en la que DTS procesó el evento (milisegundos Unix)
`id`	number	Identificador interno del evento en DTS
`pkNames`	array	Nombres de los campos de clave primaria (por lo general, `["_id"]`)
`data`	array	Datos del documento después de la operación. En las actualizaciones parciales (Oplog o ChangeStream sin recuperación del documento completo), solo contiene los campos modificados mediante operadores de actualización de MongoDB (`$set`, `$unset`). Para eventos DDL es `null`.
`old`	array	Estado del documento antes de una actualización: solo se incluye el campo `_id`. Para eventos INSERT y DDL es `null`.
`sql`	object o null	Detalles de la instrucción DDL (cuando `isDdl: true`). Para eventos DML es `null`.
`gtid`	null	No aplica para MongoDB (reservado por compatibilidad con MySQL).
`mysqlType`	null	No aplica para MongoDB.
`serverId`	null	No aplica para MongoDB.
`sqlType`	null	No aplica para MongoDB.

Escenario 1: Oplog

Establezca Migration Method en Oplog.

Source change type	Source statement	Data received by the destination topic
`insert`	`db.kafka_test.insert({"cid":"a","person":{"name":"testName","age":NumberInt(18),"skills":["database","ai"]}})`	Consulte el ejemplo siguiente
`update $set`	`db.kafka_test.update({"cid":"a"},{$set:{"person.age":NumberInt(20)}})`	Consulte el ejemplo siguiente
`update $set new field`	`db.kafka_test.update({"cid":"a"},{$set:{"salary":100}})`	Consulte el ejemplo siguiente
`update $unset (remove field)`	`db.kafka_test.update({"cid":"a"},{$unset:{"salary":1}})`	Consulte el ejemplo siguiente
`delete`	`db.kafka_test.deleteOne({"cid":"a"})`	Consulte el ejemplo siguiente
`ddl drop`	`db.kafka_test.drop()`	Consulte el ejemplo siguiente

Consulte los datos (haga clic para expandir)

{
    "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"
}

Consulte los datos (haga clic para expandir)

{
    "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"
}

Consulte los datos (haga clic para expandir)

{
    "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"
}

Consulte los datos (haga clic para expandir)

{
    "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"
}

Consulte los datos (haga clic para expandir)

{
    "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"
}

Consulte los datos (haga clic para expandir)

{
    "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"
}

Escenario 2: ChangeStream — solo campos actualizados

Establezca Migration Method en ChangeStream. Establezca Obtain the entire document after it is updated en No.

Los mensajes de inserción y eliminación son idénticos a los del escenario 1. Los mensajes de actualización contienen únicamente los campos modificados.

Consulte los datos (haga clic para expandir)

{
    "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"
}

Consulte los datos (haga clic para expandir)

{
    "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"
}

Escenario 3: ChangeStream — documento completo en la actualización

Establezca Migration Method en ChangeStream. Establezca Obtain the entire document after it is updated en Yes.

Los eventos de actualización entregan el documento completo después del cambio, en lugar de solo los campos modificados.

Consulte los datos (haga clic para expandir)

{
    "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"
}

Consulte los datos (haga clic para expandir)

{
    "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"
}

Consulte los datos (haga clic para expandir)

{
    "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"
}

Caso especial: falta fullDocument en ChangeStream

Cuando el campo fullDocument de un evento de actualización de ChangeStream está ausente (por ejemplo, cuando un documento se mueve entre shards en una colección fragmentada), el mensaje entregado recurre al comportamiento de Oplog. Es decir, se envía una actualización parcial con los operadores $set o $unset.

Ejemplo: actualización de una colección fragmentada donde falta fullDocument

Datos base de origen:

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

Cambio incremental de origen:

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"}})

Consulte los datos (haga clic para expandir)

{
    "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"
}

Preguntas frecuentes

¿Puedo cambiar el formato de compresión de datos de Kafka (Kafka Data Compression Format) después de crear la tarea?

Sí. Consulte Modify the objects to be synchronized.

¿Puedo cambiar el mecanismo de confirmación de mensajes (Message acknowledgment mechanism) después de crear la tarea?

Sí. Consulte Modify the objects to be synchronized.

Data Transmission Service:Sincronización de datos de ApsaraDB for MongoDB a Message Queue for Apache Kafka

Requisitos previos

Límites

Límites de la base de datos de origen

Otros límites

Facturación

Tipos de sincronización y operaciones compatibles

Permisos obligatorios de la cuenta de base de datos

Creación de una tarea de sincronización

Paso 1: Abrir la página Data Synchronization

Paso 2: Configurar las bases de datos de origen y destino

Paso 3: Probar la conectividad

Paso 4: Configurar los objetos que se van a sincronizar

Paso 5: Configurar los ajustes avanzados

Paso 6: Ejecutar la comprobación previa

Paso 7: Adquirir e iniciar la instancia

Configuración de la asignación de colecciones a topics

Ejemplos de entrega de datos

Elija su escenario

Campos del mensaje Canal JSON

Escenario 1: Oplog

Escenario 2: ChangeStream — solo campos actualizados

Escenario 3: ChangeStream — documento completo en la actualización

Caso especial: falta fullDocument en ChangeStream

Preguntas frecuentes

Pasos siguientes