Configurar a sincronização inicial de várias coleções

Características da sincronização inicial de várias coleções

Uma operação initialSync de várias coleções tem as seguintes características:

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

• Retoma sem ressincronizar coleções concluídas. O Atlas Stream Processing acompanha quais coleções foram concluídas, portanto, um processador de fluxo reiniciado ignora as coleções concluídas e retoma qualquer coleção em andamento a partir do seu último documento sincronizado. Ele pode reenviar documentos que já emitiu, portanto, projete seu coletor para lidar com documentos repetidos de forma idempotente.

• Expõe o progresso de sincronização granular no nível da coleção. stats().stats.operatorStats[0].targetStats reports sync progress and document counts for each target collection. To learn more, see Verify Initial Sync Progress.

Diferenças de comportamento a serem consideradas

A sincronização de várias coleções initialSync difere da sincronização de coleção única de maneiras que afetam como você configura e opera seu processador de fluxo:

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

• O escopo de catchup segue o escopo da fonte. Após a fase de sincronização, o Atlas Stream Processing abre um fluxo de alterações que corresponde ao escopo da fonte: um fluxo de alterações em nível de banco de dados para uma fonte de banco de dados inteiro, um fluxo de alterações em nível de cluster para uma fonte de cluster inteiro ou um fluxo de alterações com filtro para uma lista explícita de coleção. Você não precisa configurar isso separadamente.

• Uma falha na sincronização da coleção faz com que todo o processador falhe. Assim como na sincronização de coleção única, se o Atlas Stream Processing falhar ao sincronizar qualquer coleção da lista, todo o processador de stream falha.

• Algumas configurações do lado da origem não são replicadas para o destino. initialSync não replica índices, opções de coleção, como coleções limitadas, coleções de séries temporais, validadores ou agrupamentos 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 coleções que você cria depois que o processador de fluxo é iniciado; o conjunto de coleções a ser sincronizado é fixo nesse ponto.

Pré-requisitos

Antes de concluir este procedimento, você deve ter:

• Um espaço de trabalho de Stream Processing com conexões Atlas para o cluster de origem e de destino. Para adicionar uma conexão, consulte Gerenciar conexões.

• Um tier de processador que suporta o número de coleções que você planeja listar. Para aprender 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 seguintes etapas 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

Select the Atlas UI or mongosh to configure the stream processor.

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 para cada coleção de destino. A entrada do operador de origem em stats.operatorStats relata um 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 alterna entre pending, in_progress e completed enquanto o Atlas Stream Processing sincroniza essa coleção, ou relata failed se a sincronização falhar. estimatedDocs e copiedDocs relatam as contagens estimadas e concluídas de documentos para aquela coleção.

Para visualizar o progresso total em todas as coleções de destino, use o campo initialSync do próprio operador de origem, que soma estimatedDocs e copiedDocs em todas as coleções:

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

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