Configurar sincronização inicial de múltiplas coleções

Traços de sincronização inicial de múltiplas coleções

Uma operação multi-collection initialSync tem as seguintes características:

• Sincroniza coleções na ordem da lista. coll: ["a", "b", "c"]Com, o Atlas Stream Processing drena a as partições de antes b de e b antes c de, exceto quando o transbordamento preenche a capacidade ociosa com partições de uma collection posterior.

• Retoma sem ressincronizar as coleções concluídas. O Atlas Stream Processing acompanha quais collections foram concluídas, portanto, um stream operators reiniciado ignora as collections concluídas e retoma qualquer collection em andamento do último documento sincronizado. Ele pode devolver documentos já emitidos, portanto, projete seu coletor para lidar com documentos repetidos de forma idempotente.

• Expõe o progresso granular da sincronização no nível da collection. stats().stats.operatorStats[0].targetStats os relatórios sincronizam o progresso e as contagens de documento para cada coleção de destino. Para saber mais, consulte Verificar o progresso da sincronização inicial.

Diferenças de comportamento a serem consideradas

A multicollection initialSync difere da sincronização de collection única de maneiras que afetam o modo como você configura e opera o processador de stream:

• O paralelismo é global, não por collection. A configuração limita initialSync.parallelism o número de leituras simultâneas de partição em todo o conjunto de coleções de destino, não por coleção.

• O escopo da recuperaçãosegue o escopo da origem. Após a fase de sincronização, o Atlas Stream Processing abre um change stream que corresponde ao escopo da origem: um change stream no nível do banco de dados para uma origem de banco de dados inteiro, um change stream no nível do cluster para uma origem de todo o cluster ou um change stream filtrado para um lista explícita de coleções. Você não precisa configurar isso separadamente.

• Uma falha de sincronização da coleção falha em todo o processador. Assim como na sincronização de coleção única, se o Atlas Stream Processing não sincronizar nenhuma coleção na lista, todo o processador de fluxo falhará.

• Algumas configurações do lado da origem não são replicadas para o destino. initialSync não replica índices, opções de collection, como capped collections, coleções de séries temporais, validadores ou agrupamento padrão, ou definições de visualização. Crie-os no destino antes de iniciar o processador de fluxo. O Atlas Stream Processing também não descobre as collections que você cria após o início do processador de stream; o conjunto de coleções a serem sincronizadas é fixado nesse ponto.

Pré-requisitos

Antes de concluir este procedimento, você deve ter:

• Um espaço de trabalho de processamento de fluxo com conexões Atlas para os clusters de origem e destino. Para adicionar uma conexão, consulte Gerenciar conexões.

• Uma camada de processador que ofereça suporte ao número de coleções que você planeja listar. Para saber mais, consulte Alocação de recursos.

Execute os comandos nesta seção em seu cluster. Para se conectar, consulte Conectar-se a um cluster via mongosh.

Conclua as etapas a seguir para preparar as coleções de origem:

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

{ ok: 1, ... }

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

{ ok: 1, ... }

Procedimento

Selecione a UI do Atlas ou para configurar o processador de mongosh fluxo.

Verificar o progresso da sincronização inicial

Depois de iniciar o processador de fluxo, chame sp.<processor-name>.stats() com a opção verbose para visualizar o progresso da sincronização de cada coleção de destino. A entrada do operador de origem em stats.operatorStats relata uma array targetStats com um elemento por coleção:

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'
    }
  }
]

O campo initialSync.status de cada destino passa por pending, in_progress e completed à medida que o Atlas Stream Processing sincroniza essa coleção ou relata failed se a sincronização falhar. estimatedDocs e copiedDocs informam as contagens de documento estimadas e concluídas para essa collection.

Para visualizar o progresso total em cada collection de destino, use o campoinitialSync do próprio operador de origem, que resume estimatedDocs e copiedDocs em todas as collections:

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

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