Configurar la sincronización inicial de múltiples colecciones

Características de la sincronización inicial de múltiples colecciones

Una operación de colección múltiple initialSync tiene las siguientes características:

• Sincroniza las colecciones en el orden de la lista. Con coll: ["a", "b", "c"], Atlas Stream Processing vacía las particiones de a antes que las de b, y las de b antes que las de c, excepto cuando el desbordamiento llena la capacidad inactiva con particiones de una colección posterior.

• Se reanuda sin resincronizar las colecciones completadas. Atlas Stream Processing realiza un seguimiento de las colecciones que han finalizado, por lo que un procesador de flujo reiniciado omite las colecciones completadas y reanuda cualquier colección en curso desde su último documento sincronizado. Podría volver a entregar documentos que ya emitió, así que diseñe su receptor para que gestione los documentos repetidos de forma idempotente.

• Muestra el progreso de sincronización a nivel de colección con granularidad. stats().stats.operatorStats[0].targetStats Los informes muestran el progreso de la sincronización y el recuento de documentos para cada colección de destino. Para obtener más información, consulte Verificar el progreso de la sincronización inicial.

Diferencias de comportamiento a considerar

La sincronización de varias colecciones initialSync difiere de la sincronización de una sola colección de maneras que afectan cómo configura y opera su procesador de transmisiones:

• El paralelismo es global, no por colección. Establecer initialSync.parallelism limita el número de lecturas de particiones simultáneas en todo el conjunto de colecciones de destino, no por colección.

• El ámbito de recuperación sigue al ámbito de origen. Tras la fase de sincronización, Atlas Stream Processing abre un flujo de cambios que coincide con el ámbito de origen: un flujo de cambios a nivel de base de datos para un origen de base de datos completa, un flujo de cambios a nivel de clúster para un origen de clúster completo o un flujo de cambios filtrado para una lista explícita de colecciones. No es necesario configurarlo por separado.

• Un fallo de sincronización de la colección provoca el fallo de todo el procesador. Al igual que con la sincronización de una sola colección, si Atlas Stream Processing no puede sincronizar cualquier colección de la lista, todo el procesador de transmisión falla.

• Algunos ajustes de la configuración de origen no se replican en el destino. initialSync no replica índices, opciones de colección como colecciones con tamaño fijo, colecciones de series de tiempo, validadores, intercalación por defecto o definiciones de vista. Cree estos en el destino antes de iniciar el procesador de transmisión. Atlas Stream Processing tampoco detecta las colecciones que cree después de que se inicie el procesador de transmisión; el conjunto de colecciones a sincronizar se fija en ese punto.

Requisitos previos

Antes de completar este procedimiento, debe tener:

• Un espacio de trabajo de Stream Processing con conexiones de Atlas a los clúster de origen y de destino. Para añadir una conexión, consulte Gestionar conexiones.

• Un nivel de procesador que admita el número de colecciones que planea enumerar. Para aprender más información, consulte Asignación de recursos.

Ejecute los comandos de esta sección en su clúster. Para conectarse, consulte Conectarse a un clúster mediante mongosh.

Siga los pasos siguientes para preparar la colección de origen:

db.getSiblingDB("sample_analytics").runCommand({
   collMod: "customers",
   changeStreamPreAndPostImages: { enabled: true }
})

{ ok: 1, ... }

db.getSiblingDB("sample_analytics").runCommand({
   collMod: "accounts",
   changeStreamPreAndPostImages: { enabled: true }
})

{ ok: 1, ... }

Procedimiento

Seleccione la interfaz de usuario de Atlas o para configurar el procesador de flujo.mongosh

Verificar el progreso de la sincronización inicial

Una vez que inicie el procesador de transmisiones, llame a sp.<processor-name>.stats() con la opción verbose para ver el progreso de sincronización para cada colección de destino. La entrada del operador de origen en stats.operatorStats informa un arreglo targetStats con un elemento por colección:

sp.replicate_analytics_sp.stats(
  { options: { verbose: true } }
).stats.operatorStats[0].targetStats

[
  {
    db: 'sample_analytics',
    coll: 'customers',
    inputMessageCount: Long('500'),
    inputMessageSize: 672326,
    initialSync: {
      estimatedDocs: Long('500'),
      copiedDocs: Long('251'),
      status: 'in_progress'
    }
  },
  {
    db: 'sample_analytics',
    coll: 'accounts',
    inputMessageCount: Long('1746'),
    inputMessageSize: 1883681,
    initialSync: {
      estimatedDocs: Long('1746'),
      copiedDocs: Long('1746'),
      status: 'completed'
    }
  }
]

El campo initialSync.status de cada destino pasa por pending, in_progress y completed a medida que Atlas Stream Processing sincroniza esa colección, o informa failed si la sincronización falla. estimatedDocs y copiedDocs informan los recuentos de documentos estimados y completados para esa colección.

Para ver el progreso total en todas las colecciones de destino, utilice el propio campo initialSync del operador de origen, que suma estimatedDocs y copiedDocs en todas las colecciones:

sp.replicate_analytics_sp.stats(
  { options: { verbose: true } }
).stats.operatorStats[0].initialSync

{ estimatedDocs: Long('2246'), copiedDocs: Long('1997') }