MongoDB - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

El conector de MongoDB integra ApsaraDB for MongoDB y las instalaciones de MongoDB autogestionadas en Realtime Compute for Apache Flink como tablas de origen, de dimensiones y de destino. Utiliza la API de Change Stream para capturar eventos de inserción, actualización, reemplazo y eliminación en tiempo real.

Capacidades

Categoría	Descripción
Tipos de tabla	Origen SQL, búsqueda (dimensiones) y destino Origen de Flink CDC Origen de DataStream
Modo de ejecución	Streaming
Tipos de API	API de DataStream, SQL, Flink CDC
Semántica de escritura del destino	Inserción, actualización y eliminación (con clave primaria declarada)

Métricas de monitoreo

Tablas de origen:

numBytesIn, numBytesInPerSecond, numRecordsIn, numRecordsInPerSecond, numRecordsInErrors, currentFetchEventTimeLag, currentEmitEventTimeLag, watermarkLag, sourceIdleTime

Las tablas de dimensiones y de destino no exponen métricas de monitoreo.

Para consultar las definiciones de las métricas, consulte Metrics.

Funcionamiento

El conector de MongoDB lee los datos en dos fases:

Instantánea completa: lee en paralelo todos los documentos existentes de las colecciones indicadas.
Lectura incremental: cambia automáticamente al consumo del oplog mediante la API de Change Stream una vez que termina la instantánea.

Este proceso ofrece semántica exactly-once. De este modo, se garantiza que no haya registros duplicados ni faltantes durante la recuperación ante fallos.

Conceptos clave

Modos de inicio

Elija un modo de inicio según el momento en que su canalización deba empezar a consumir datos:

Mode	Behavior	Use when
`initial` (predeterminado)	Lee una instantánea en el primer inicio y luego cambia a la lectura incremental	Cuando necesita una copia completa de los datos existentes
`latest-offset`	Inicia desde la posición actual del oplog; no incluye datos históricos	Si solo requiere los cambios a partir de este momento
`timestamp`	Lee los eventos del oplog desde una marca de tiempo específica; omite la instantánea	Más indicado para escenarios donde se necesitan los cambios desde un punto conocido en el tiempo (requiere MongoDB 4.0 o superior)

Compatibilidad con el registro de cambios completo

De forma predeterminada, MongoDB no almacena el estado previo de los documentos antes de un cambio (en versiones anteriores a MongoDB 6.0). Sin esta información, el conector solo genera eventos UPSERT, por lo que faltan los registros UPDATE_BEFORE.

Para solucionar esto, el planificador de Flink SQL inserta un operador ChangelogNormalize que almacena en caché el estado de los documentos en el backend de estado de Flink. Aunque funciona, este enfoque consume una cantidad considerable de almacenamiento de estado.

MongoDB 6.0 o superior admite el registro de imágenes previas y posteriores. Al habilitar esta función, MongoDB registra el estado completo del documento antes y después de cada cambio. Si establece scan.full-changelog en true, el conector usará estos registros para generar flujos completos de registro de cambios. Esto elimina el operador ChangelogNormalize y su sobrecarga de estado.

Requisitos previos

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

Una instancia de ApsaraDB for MongoDB (conjunto de réplicas o clúster fragmentado), o bien un clúster de MongoDB 3.6 o superior autogestionado con el modo de conjunto de réplicas habilitado. Consulte Replicación.
Si la autenticación está habilitada, un usuario de MongoDB con los siguientes permisos: splitVector, listDatabases, listCollections, collStats, find, changeStream y acceso de lectura a config.collections y config.chunks.
Las direcciones IP del clúster de Flink agregadas a la lista de permitidos de IP de MongoDB.
La base de datos y la colección de destino creadas antes de ejecutar el trabajo.

Limitaciones

Origen SQL

La lectura paralela de instantáneas requiere MongoDB 4.0 o posterior. Habilítela si establece scan.incremental.snapshot.enabled en true.
No es posible monitorear las bases de datos admin, local y config, ni las colecciones del sistema. Esta es una restricción de Change Stream de MongoDB. Consulte Change Streams en la documentación de MongoDB.
Al crear una tabla de origen SQL, declare la columna _id STRING y establézcala como clave primaria.

Destino SQL

VVR 8.0.4 y anteriores: solo inserción.
VVR 8.0.5 y posteriores con clave primaria declarada: inserción, actualización y eliminación.
VVR 8.0.5 y posteriores sin clave primaria: solo inserción.
No se admite la entrega exactly-once. La opción sink.delivery-guarantee acepta none o at-least-once.

Búsqueda SQL (dimensiones)

Compatible con VVR 8.0.5 y posteriores.
VVR 8.0.9 y posteriores: las uniones de búsqueda admiten la lectura del campo integrado _id de tipo ObjectId.

SQL

Sintaxis

CREATE TABLE tableName(
  _id STRING,
  [columnName dataType,]*
  PRIMARY KEY(_id) NOT ENFORCED
) WITH (
  'connector' = 'mongodb',
  'hosts' = 'localhost:27017',
  'username' = 'mongouser',
  'password' = '${secret_values.password}',
  'database' = 'testdb',
  'collection' = 'testcoll'
)

Declare la columna _id STRING y especifíquela como clave primaria al crear una tabla de origen CDC.

Opciones del conector

Generales

Opción	Tipo	Obligatorio	Predeterminado	Descripción
`connector`	String	Sí	—	Identificador del conector. Para tablas de origen: `mongodb-cdc` (VVR 8.0.4 y anteriores) o `mongodb` / `mongodb-cdc` (VVR 8.0.5 y posteriores). Para tablas de dimensiones o destino: `mongodb`.
`uri`	String	No	—	URI de conexión de MongoDB. Especifique `uri` o bien `hosts`. Si especifica `uri`, omita `scheme`, `hosts`, `username`, `password` y `connection.options`. Si configura ambos parámetros, `uri` tiene prioridad.
`hosts`	String	No	—	Nombre de host del servidor de MongoDB. Separe varios hosts con comas (`,`).
`scheme`	String	No	`mongodb`	Protocolo de conexión. Valores válidos: `mongodb` (predeterminado), `mongodb+srv` (DNS SRV).
`username`	String	No	—	Nombre de usuario de MongoDB. Obligatorio cuando la autenticación está habilitada.
`password`	String	No	—	Contraseña de MongoDB. Obligatoria cuando la autenticación está habilitada. Utilice variables en lugar de codificar las credenciales directamente.
`database`	String	No	—	Nombre de la base de datos de MongoDB. Admite expresiones regulares para las tablas de origen. Si no se establece, se supervisan todas las bases de datos. No es posible supervisar las bases de datos `admin`, `local` ni `config`.
`collection`	String	No	—	Nombre de la colección de MongoDB. Admite expresiones regulares para las tablas de origen. Notas: Si no se establece, se supervisan todas las colecciones. No se pueden supervisar las colecciones del sistema. Si el nombre de la colección contiene caracteres especiales de expresiones regulares, utilice el espacio de nombres completo (`database.collection`).
`connection.options`	String	No	—	Opciones de conexión adicionales como pares `key=value` separados por `\&` (por ejemplo, `connectTimeoutMS=12000\&socketTimeoutMS=13000`). De forma predeterminada, el conector no establece un tiempo de espera para la conexión de socket. Esto puede provocar interrupciones prolongadas durante fluctuaciones de red. Establezca `socketTimeoutMS` con un valor adecuado para evitarlo.

Origen

Opción	Tipo	Obligatorio	Predeterminado	Descripción
`scan.startup.mode`	String	No	`initial`	Modo de inicio. Valores válidos: `initial`, `latest-offset`, `timestamp`. Consulte Startup modes y Startup Properties.
`scan.startup.timestamp-millis`	Long	Condicional	—	Marca de tiempo de inicio en milisegundos desde la época UNIX. Obligatorio cuando `scan.startup.mode` es `timestamp`.
`initial.snapshotting.queue.size`	Integer	No	`10240`	Tamaño máximo de la cola durante la fase inicial de instantánea. Solo surte efecto cuando `scan.startup.mode` es `initial`.
`batch.size`	Integer	No	`1024`	Tamaño del lote del cursor.
`poll.max.batch.size`	Integer	No	`1024`	Número máximo de documentos de cambio extraídos por lote durante la lectura de flujo. Los valores más altos asignan un búfer interno mayor.
`poll.await.time.ms`	Integer	No	`1000`	Intervalo entre solicitudes de extracción de datos, en milisegundos.
`heartbeat.interval.ms`	Integer	No	`0`	Intervalo de latido en milisegundos. El conector envía latidos para rastrear la posición más reciente del oplog. Si establece este valor en `0`, se deshabilitan los latidos. Configure este parámetro para colecciones que se actualizan con poca frecuencia.
`scan.incremental.snapshot.enabled`	Boolean	No	`false`	Habilita la lectura paralela de instantáneas. Función experimental. Requiere MongoDB 4.0 o posterior.
`scan.incremental.snapshot.chunk.size.mb`	Integer	No	`64`	Tamaño del fragmento para la lectura paralela de instantáneas, en MB. Función experimental. Solo surte efecto cuando la lectura paralela de instantáneas está habilitada.
`scan.full-changelog`	Boolean	No	`false`	Genera un flujo de registro de cambios completo mediante los registros de preimagen y posimagen de MongoDB. Función experimental. Requiere MongoDB 6.0 o posterior con las funciones de preimagen y posimagen habilitadas.
`scan.flatten-nested-columns.enabled`	Boolean	No	`false`	Analiza sintácticamente los campos separados por `.` como campos de documento BSON anidados. Por ejemplo, `{"nested":{"col":true}}` se asigna a un campo llamado `nested.col`. Compatible con VVR 8.0.5 y versiones posteriores.
`scan.primitive-as-string`	Boolean	No	`false`	Analiza sintácticamente todos los tipos BSON primitivos como STRING. Compatible con VVR 8.0.5 y versiones posteriores.
`scan.ignore-delete.enabled`	Boolean	No	`false`	Ignora todos los eventos DELETE (-D), incluidos los generados durante el archivado de datos de MongoDB. Compatible con VVR 11.1 y versiones posteriores.
`scan.incremental.snapshot.backfill.skip`	Boolean	No	`false`	Valores válidos: true: omite el relleno durante la lectura de instantáneas incrementales. false (predeterminado): no omite el relleno. El relleno solo se aplica durante la consulta de instantánea de un único fragmento y no abarca toda la fase de lectura completa. Cuando se omite el relleno, la consulta de instantánea de cada fragmento lee los datos más recientes en ese instante. Las actualizaciones que ocurren en un fragmento después de su lectura no se combinan durante la fase de lectura completa. En su lugar, se leen del OpLog tras entrar en la fase incremental. Por ejemplo, una actualización en chunk5 que ocurre mientras se toma la instantánea de chunk5 se refleja directamente en dicha instantánea. Sin embargo, si chunk5 se actualiza después de que el lector avanzó a chunk80, la actualización se aplica más tarde desde el OpLog durante la fase incremental. Importante Cuando esta opción está habilitada, los cambios que ocurren durante o después del análisis de un fragmento se entregan desde el OpLog en la fase incremental y podrían duplicarse. Solo se garantiza la semántica de al menos una vez. Habilítela únicamente cuando el destino descendente admita escrituras idempotentes por clave principal. Nota Compatible solo con VVR 11.1 y versiones posteriores.
`initial.snapshotting.pipeline`	String	No	—	Operaciones de canalización de agregación de MongoDB aplicadas durante la lectura de instantáneas para filtrar datos. Especifíquelas como una matriz JSON, por ejemplo: `[{"$match": {"closed": "false"}}]`. Solo surte efecto cuando `scan.startup.mode` es `initial` y el conector se ejecuta en modo Debezium. Compatible con VVR 11.1 y versiones posteriores.
`initial.snapshotting.max.threads`	Integer	No	—	Número de subprocesos para la replicación de instantáneas. Solo surte efecto cuando `scan.startup.mode` es `initial`. Compatible con VVR 11.1 y versiones posteriores.
`initial.snapshotting.queue.size`	Integer	No	`16000`	Tamaño de la cola para la instantánea inicial. Solo surte efecto cuando `scan.startup.mode` es `initial`. Compatible con VVR 11.1 y versiones posteriores.
`scan.change-stream.reading.parallelism`	Integer	No	`1`	Número de lectores simultáneos para Change Stream. Solo surte efecto cuando `scan.incremental.snapshot.enabled` es `true`. Establezca también `heartbeat.interval.ms` cuando utilice esta opción. Compatible con VVR 11.2 y versiones posteriores.
`scan.change-stream.reading.queue-size`	Integer	No	`16384`	Tamaño de la cola de mensajes para suscripciones simultáneas de change stream. Solo surte efecto cuando `scan.change-stream.reading.parallelism` está habilitado. Compatible con VVR 11.2 y versiones posteriores.

Lookup (dimensiones)

Opción	Tipo	Obligatorio	Predeterminado	Descripción
`lookup.cache`	String	No	`NONE`	Política de caché. Valores válidos: `NONE` (sin caché), `PARTIAL` (almacena en caché los resultados de búsqueda de la base de datos externa).
`lookup.max-retries`	Integer	No	`3`	Número máximo de reintentos cuando falla una búsqueda.
`lookup.retry.interval`	Duration	No	`1s`	Intervalo entre reintentos cuando falla una búsqueda.
`lookup.partial-cache.expire-after-access`	Duration	No	—	Tiempo máximo que permanece una entrada en caché después del último acceso. Unidades compatibles: `ms`, `s`, `min`, `h`, `d`. Requiere que `lookup.cache` sea `PARTIAL`.
`lookup.partial-cache.expire-after-write`	Duration	No	—	Tiempo máximo que permanece una entrada en caché después de escribirse. Requiere que `lookup.cache` sea `PARTIAL`.
`lookup.partial-cache.max-rows`	Long	No	—	Número máximo de filas en la caché. Cuando se alcanza el límite, se eliminan las entradas más antiguas. Requiere que `lookup.cache` sea `PARTIAL`.
`lookup.partial-cache.cache-missing-key`	Boolean	No	`true`	Almacena una entrada nula en caché cuando una clave de búsqueda no tiene ningún registro coincidente. Requiere que `lookup.cache` sea `PARTIAL`.
`lookup.type-conversion.mode`	String	No	`FORCE_STRING`	Estrategia de conversión utilizada cuando los valores de cadena de Flink se comparan con la tabla de dimensiones de MongoDB. Valores válidos: `FORCE_STRING`: siempre convierte los campos de MongoDB a cadenas antes de la comparación. Permite comparar tipos de campo sin equivalente en Flink, como ObjectId, pero no utiliza el índice del campo (si existe). `OBJECT_ID_WRAPPER`: si un valor de cadena de Flink es una cadena hexadecimal válida de ObjectId, lo compara como ObjectId. Se puede usar el índice del campo de unión para acelerar la consulta. `NONE`: no aplica ningún manejo especial. No coincide ninguna fila si el tipo de MongoDB difiere del tipo de consulta de Flink. Nota Este parámetro solo es compatible con VVR 11.9 y versiones posteriores.

Destino

Option	Type	Required	Default	Description
`sink.buffer-flush.max-rows`	Integer	No	`1000`	Número máximo de registros escritos por lote.
`sink.buffer-flush.interval`	Duration	No	`1s`	Intervalo de vaciado.
`sink.delivery-guarantee`	String	No	`at-least-once`	Semántica de entrega de escritura. Valores válidos: `none`, `at-least-once`. No se admite exactamente una vez.
`sink.max-retries`	Integer	No	`3`	Número máximo de reintentos cuando falla una escritura.
`sink.retry.interval`	Duration	No	`1s`	Intervalo entre reintentos cuando falla una escritura.
`sink.parallelism`	Integer	No	—	Paralelismo personalizado del destino.
`sink.delete-strategy`	String	No	`CHANGELOG_STANDARD`	Estrategia para manejar eventos -D y -U. Valores válidos: `CHANGELOG_STANDARD` (aplica actualizaciones y eliminaciones de forma normal), `IGNORE_DELETE` (ignora los eventos -D y sobrescribe filas completas en -U), `PARTIAL_UPDATE` (ignora los eventos -U para admitir actualizaciones parciales de columnas y elimina filas en -D), `IGNORE_ALL` (ignora tanto los eventos -U como -D).

Asignaciones de tipos de datos

Origen

BSON type	Flink SQL
Int32	INT
Int64	BIGINT
Double	DOUBLE
Decimal128	DECIMAL(p, s)
Boolean	BOOLEAN
Date Timestamp	DATE
Date Timestamp	TIME
DateTime	TIMESTAMP(3), TIMESTAMP_LTZ(3)
Timestamp	TIMESTAMP(0), TIMESTAMP_LTZ(0)
String, ObjectId, UUID, Symbol, MD5, JavaScript, Regex	STRING
Binary	BYTES
Object	ROW
Array	ARRAY
DBPointer	ROW\<$ref STRING, $id STRING\>
GeoJSON Point	ROW\<type STRING, coordinates ARRAY\<DOUBLE\>\>
GeoJSON Line	ROW\<type STRING, coordinates ARRAY\<ARRAY\<DOUBLE\>\>\>

Lookup (dimensiones) y destino

BSON type	Flink SQL type
Int32	INT
Int64	BIGINT
Double	DOUBLE
Decimal128	DECIMAL
Boolean	BOOLEAN
DateTime	TIMESTAMP_LTZ(3)
Timestamp	TIMESTAMP_LTZ(0)
String, ObjectId	STRING
Binary	BYTES
Object	ROW
Array	ARRAY

Columnas de metadatos

El origen SQL admite a continuación columnas de metadatos:

Metadata column	Type	Description
`database_name`	STRING NOT NULL	Base de datos que contiene el documento.
`collection_name`	STRING NOT NULL	Colección que contiene el documento.
`op_ts`	TIMESTAMP_LTZ(3) NOT NULL	Hora en la que cambió el documento. Devuelve `0` para los documentos de la instantánea inicial.
`row_kind`	STRING NOT NULL	Tipo de evento de cambio: `+I` (INSERT), `-D` (DELETE), `-U` (UPDATE_BEFORE), `+U` (UPDATE_AFTER). Compatible con VVR 11.1 y versiones posteriores.

Ejemplos

Origen

En el siguiente ejemplo se lee de una tabla de origen de MongoDB con lectura paralela de instantáneas y registro de cambios completo habilitados. Luego, se escriben los campos seleccionados en un destino de impresión.

-- CDC source table: reads product data from MongoDB
-- _id must be declared and set as the primary key
CREATE TEMPORARY TABLE mongo_source (
  `_id` STRING,
  name STRING,
  weight DECIMAL,
  tags ARRAY<STRING>,
  price ROW<amount DECIMAL, currency STRING>,
  suppliers ARRAY<ROW<name STRING, address STRING>>,
  db_name STRING METADATA FROM 'database_name' VIRTUAL,
  collection_name STRING METADATA VIRTUAL,
  op_ts TIMESTAMP_LTZ(3) METADATA VIRTUAL,
  PRIMARY KEY(_id) NOT ENFORCED
) WITH (
  'connector' = 'mongodb',
  'hosts' = 'dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,',
  'username' = 'root',
  'password' = '${secret_values.password}',
  'database' = 'flinktest',
  'collection' = 'flinkcollection',
  'scan.incremental.snapshot.enabled' = 'true', -- Enable parallel snapshot reading (requires MongoDB 4.0+)
  'scan.full-changelog' = 'true'                -- Enable full changelog (requires MongoDB 6.0+ with preimage/postimage)
);

CREATE TEMPORARY TABLE productssink (
  name STRING,
  weight DECIMAL,
  tags ARRAY<STRING>,
  price_amount DECIMAL,
  suppliers_name STRING,
  db_name STRING,
  collection_name STRING,
  op_ts TIMESTAMP_LTZ(3)
) WITH (
  'connector' = 'print',
  'logger' = 'true'
);

INSERT INTO productssink
SELECT
  name,
  weight,
  tags,
  price.amount,
  suppliers[1].name,
  db_name,
  collection_name,
  op_ts
FROM mongo_source;

Lookup (dimensiones)

En el siguiente ejemplo se une un flujo de generador de datos con una tabla de dimensiones de MongoDB mediante una unión temporal.

CREATE TEMPORARY TABLE datagen_source (
  id STRING,
  a INT,
  b BIGINT,
  `proctime` AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE mongo_dim (
  `_id` STRING,
  name STRING,
  weight DECIMAL,
  tags ARRAY<STRING>,
  price ROW<amount DECIMAL, currency STRING>,
  suppliers ARRAY<ROW<name STRING, address STRING>>,
  PRIMARY KEY(_id) NOT ENFORCED
) WITH (
  'connector' = 'mongodb',
  'hosts' = 'dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,',
  'username' = 'root',
  'password' = '${secret_values.password}',
  'database' = 'flinktest',
  'collection' = 'flinkcollection',
  'lookup.cache' = 'PARTIAL',                           -- Cache lookup results for better performance
  'lookup.partial-cache.expire-after-access' = '10min', -- Evict cached entries after 10 minutes of inactivity
  'lookup.partial-cache.expire-after-write' = '10min',  -- Evict cached entries 10 minutes after write
  'lookup.partial-cache.max-rows' = '100'               -- Maximum 100 rows in the cache
);

CREATE TEMPORARY TABLE print_sink (
  name STRING,
  weight DECIMAL,
  tags ARRAY<STRING>,
  price_amount DECIMAL,
  suppliers_name STRING
) WITH (
  'connector' = 'print',
  'logger' = 'true'
);

INSERT INTO print_sink
SELECT
  T.id,
  T.a,
  T.b,
  H.name
FROM datagen_source AS T
JOIN mongo_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H ON T.id = H._id;

Destino

En el siguiente ejemplo se escriben datos de un generador de datos en una tabla de destino de MongoDB. Se declara una clave principal para admitir operaciones de inserción, actualización y eliminación.

CREATE TEMPORARY TABLE datagen_source (
  `_id` STRING,
  name STRING,
  weight DECIMAL,
  tags ARRAY<STRING>,
  price ROW<amount DECIMAL, currency STRING>,
  suppliers ARRAY<ROW<name STRING, address STRING>>
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE mongo_sink (
  `_id` STRING,
  name STRING,
  weight DECIMAL,
  tags ARRAY<STRING>,
  price ROW<amount DECIMAL, currency STRING>,
  suppliers ARRAY<ROW<name STRING, address STRING>>,
  PRIMARY KEY(_id) NOT ENFORCED -- Declare primary key to enable update and delete
) WITH (
  'connector' = 'mongodb',
  'hosts' = 'dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,',
  'username' = 'root',
  'password' = '${secret_values.password}',
  'database' = 'flinktest',
  'collection' = 'flinkcollection'
);

INSERT INTO mongo_sink SELECT * FROM datagen_source;

Flink CDC (vista previa pública)

Flink CDC permite sincronizar datos de MongoDB con un almacén descendente mediante una canalización basada en scripts YAML, sin necesidad de escribir DDL de SQL. Esta función requiere VVR 11.1 o posterior.

Sintaxis

source:
   type: mongodb
   name: MongoDB Source
   hosts: localhost:33076
   username: ${mongo.username}
   password: ${mongo.password}
   database: foo_db
   collection: foo_col_.*

sink:
  type: ...

Opciones de configuración

Opción	Obligatorio	Tipo	Predeterminado	Descripción
`type`	Sí	STRING	—	El conector. Establezca el valor en `mongodb`.
`scheme`	No	STRING	`mongodb`	Protocolo de conexión. Valores válidos: `mongodb`, `mongodb+srv`.
`hosts`	Sí	STRING	—	Nombres de host del servidor de MongoDB. Separe varios hosts con comas.
`username`	No	STRING	—	Nombre de usuario de MongoDB.
`password`	No	STRING	—	Contraseña de MongoDB.
`database`	Sí	STRING	—	Nombre de la base de datos de MongoDB que se va a capturar. Es compatible con expresiones regulares.
`collection`	Sí	STRING	—	Nombre de la colección de MongoDB que se va a capturar. Es compatible con expresiones regulares. Utilice el espacio de nombres completo `database.collection`.
`connection.options`	No	STRING	—	Opciones de conexión adicionales como pares `k=v` separados por `\&`. Por ejemplo: `replicaSet=test\&connectTimeoutMS=300000`.
`schema.inference.strategy`	No	STRING	`continuous`	Estrategia de inference de esquema. `continuous`: deduce los tipos de forma continua y emite eventos de cambio de esquema cuando este se amplía. `static`: deduce el esquema una sola vez durante el inicio.
`scan.max.pre.fetch.records`	No	INT	`50`	Número máximo de registros que se muestrean por colección durante la inference inicial del esquema.
`scan.startup.mode`	No	STRING	`initial`	Modo de inicio. Valores válidos: `initial`, `latest-offset`, `timestamp`, `snapshot`.
`scan.startup.timestamp-millis`	No	LONG	—	Marca de tiempo de inicio en milisegundos. Es obligatorio cuando `scan.startup.mode` es `timestamp`.
`chunk-meta.group.size`	No	INT	`1000`	Tamaño máximo del fragmento de metadatos.
`scan.incremental.close-idle-reader.enabled`	No	BOOLEAN	`false`	Cierra los lectores de origen inactivos después de cambiar a la lectura incremental.
`scan.incremental.snapshot.backfill.skip`	No	BOOLEAN	`false`	Valores válidos: true: omite el backfill durante la lectura de la instantánea incremental. false (predeterminado): no omite el backfill. El backfill solo se aplica durante la consulta de instantánea de un único fragmento y no abarca toda la fase de lectura completa. Cuando se omite el backfill, la consulta de instantánea de cada fragmento lee los datos más recientes en ese instante. Las actualizaciones que ocurren en un fragmento después de leerlo no se combinan durante la fase de lectura completa. En su lugar, se leen desde el OpLog tras entrar en la fase incremental. Por ejemplo, si se actualiza chunk5 mientras se toma su instantánea, el cambio se refleja directamente en ella. Sin embargo, si chunk5 se actualiza después de que el lector avanzó a chunk80, el cambio se aplica más tarde desde el OpLog durante la fase incremental. Importante Cuando esta opción está habilitada, los cambios que ocurren durante o después del análisis de un fragmento se entregan igualmente desde el OpLog en la fase incremental y podrían duplicarse. Solo se garantiza la semántica at-least-once. Habilítela únicamente cuando el destino descendente admita escrituras idempotentes por clave principal.
`scan.incremental.snapshot.unbounded-chunk-first.enabled`	No	BOOLEAN	`false`	Lee primero los fragmentos ilimitados. Reduce el riesgo de agotar la memoria en colecciones que se actualizan con frecuencia.
`batch.size`	No	INT	`1024`	Tamaño del lote del cursor.
`poll.max.batch.size`	No	INT	`1024`	Número máximo de entradas por solicitud de extracción de Change Stream.
`poll.await.time.ms`	No	INT	`1000`	Tiempo mínimo de espera entre solicitudes de extracción de Change Stream, en milisegundos.
`heartbeat.interval.ms`	No	INT	`0`	Intervalo de heartbeat en milisegundos. Configure este valor para colecciones que se actualizan con poca frecuencia. Si lo establece en `0`, se deshabilitan los heartbeats.
`scan.incremental.snapshot.chunk.size.mb`	No	INT	`64`	Tamaño del fragmento durante la creación de la instantánea, en MB.
`scan.incremental.snapshot.chunk.samples`	No	INT	`20`	Número de muestras utilizadas para estimar el tamaño de la colección durante la creación de la instantánea.
`scan.full-changelog`	No	BOOLEAN	`false`	Genera eventos de registro de cambios completos mediante registros preimage y postimage. Requiere MongoDB 6.0 o posterior con las opciones preimage y postimage habilitadas.
`scan.cursor.no-timeout`	No	BOOLEAN	`false`	Deshabilita el tiempo de espera del cursor. De forma predeterminada, MongoDB cierra los cursores inactivos después de 10 minutos.
`scan.ignore-delete.enabled`	No	BOOLEAN	`false`	Ignora los eventos de eliminación provenientes de MongoDB.
`scan.flatten.nested-documents.enabled`	No	BOOLEAN	`false`	Aplana los documentos BSON anidados. Por ejemplo, `{"doc": {"foo": 1, "bar": "two"}}` se convierte en `doc.foo INT, doc.bar STRING`.
`scan.all.primitives.as-string.enabled`	No	BOOLEAN	`false`	Deduce todos los tipos primitivos como STRING. Reduce los eventos de cambio de esquema cuando los tipos del origen son inconsistentes.
`metadata.list`	No	STRING	—	Lista separada por comas de los campos de metadatos que se envían al destino descendente. Valores compatibles: `ts_ms` (marca de tiempo del evento de OpLog), `op_ts` (alias de `ts_ms`; utilícelo al escribir metadatos en JSON de Kafka).

Asignaciones de tipos de datos

MongoDB BSON	Flink CDC	Notes
STRING	VARCHAR	—
INT32	INT	—
INT64	BIGINT	—
DECIMAL128	DECIMAL	—
DOUBLE	DOUBLE	—
BOOLEAN	BOOLEAN	—
TIMESTAMP	TIMESTAMP	—
DATETIME	LOCALZONEDTIMESTAMP	—
BINARY	VARBINARY	—
DOCUMENT	MAP	Los tipos de clave y valor se deducen automáticamente.
ARRAY	ARRAY	El tipo de los elementos se deduce automáticamente.
OBJECTID	VARCHAR	Se representa como una cadena hexadecimal.
SYMBOL, REGULAREXPRESSION, JAVASCRIPT, JAVASCRIPTWITHSCOPE	VARCHAR	Se representa como una cadena.

Columnas de metadatos

Flink CDC admite la siguiente columna de metadatos para el conector de MongoDB:

Metadata column	Type	Description
`ts_ms`	BIGINT NOT NULL	Hora en la que cambió el documento (marca de tiempo de OpLog). Devuelve `0` para los documentos de la instantánea inicial.

Utilice las columnas de metadatos genéricas del módulo Transform para acceder a database_name, collection_name y row_kind.

API de DataStream

Importante

Para usar la API de DataStream, configure el conector de DataStream en su trabajo. Consulte DataStream connector usage.

Agregar la dependencia de Maven

El repositorio central de Maven aloja los conectores de VVR para MongoDB.

<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>flink-connector-mongodb</artifactId>
    <version>${vvr.version}</version>
</dependency>

Compilar un objeto MongoDBSource

Utilice MongoDBSource.builder() para construir un origen:

Si desea habilitar la lectura de instantáneas incrementales, use el builder de com.ververica.cdc.connectors.mongodb.source.
En caso contrario, use el builder de com.ververica.cdc.connectors.mongodb.

MongoDBSource.builder()
  .hosts("mongo.example.com:27017")
  .username("mongouser")
  .password("mongopasswd")
  .databaseList("testdb")          // Supports regular expressions; use .* to match all databases
  .collectionList("testcoll")      // Supports regular expressions; use .* to match all collections
  .startupOptions(StartupOptions.initial()) // StartupOptions.latest-offset(), StartupOptions.timestamp()
  .deserializer(new JsonDebeziumDeserializationSchema())
  .build();

Parámetros de MongoDBSource

Parameter	Description
`hosts`	Nombre de host del servidor de MongoDB.
`username`	Nombre de usuario de MongoDB. Omítalo si la autenticación no está habilitada.
`password`	Contraseña de MongoDB. Omítala si la autenticación no está habilitada.
`databaseList`	Nombre de la base de datos que se va a supervisar. Es compatible con expresiones regulares. Use `.*` para coincidir con todas las bases de datos.
`collectionList`	Nombre de la colección que se va a supervisar. Es compatible con expresiones regulares. Use `.*` para coincidir con todas las colecciones.
`startupOptions`	Modo de inicio. Valores válidos: `StartupOptions.initial()`, `StartupOptions.latest-offset()`, `StartupOptions.timestamp()`.
`deserializer`	Deserializador para convertir objetos `SourceRecord`. Valores válidos: `MongoDBConnectorDeserializationSchema` (modo upsert, genera `RowData` de Flink), `MongoDBConnectorFullChangelogDeserializationSchema` (modo de registro de cambios completo, genera `RowData` de Flink), `JsonDebeziumDeserializationSchema` (genera cadenas JSON).

Referencias

Flink CDC (public preview): sincronice los datos y los cambios de esquema de MongoDB con las tablas descendentes (VVR 11.1 y posteriores).
Metrics: supervise el rendimiento de la tabla de origen.