Data Transmission Service:Sincronizar dados do ApsaraDB for MongoDB para o Message Queue for Apache Kafka

Limites do banco de dados source

• O servidor source deve ter largura de banda de saída suficiente. Largura de banda insuficiente reduz a velocidade de sincronização.

• Ao configurar o mapeamento de nomes para coleções, uma única tarefa pode sincronizar até 1.000 coleções. Exceder esse limite causa um erro de solicitação. Para sincronizar mais de 1.000 coleções, crie várias tarefas ou sincronize todo o banco de dados sem mapeamento de nomes.

• Para sources do tipo sharded cluster: o campo _id em cada coleção sincronizada deve ser único; caso contrário, pode ocorrer inconsistência de dados.

• Para sources do tipo sharded cluster: o número de nós mongos não pode exceder 10, e a instância não deve conter orphaned documents. Consulte o tópico de FAQ para saber como remover documentos órfãos.

• Instâncias independentes do ApsaraDB for MongoDB, clusters do Azure Cosmos DB for MongoDB e elastic clusters do Amazon DocumentDB não são suportados como sources.

• O DTS não consegue se conectar ao MongoDB por meio de um endpoint SRV.

• O oplog deve estar ativado e reter dados de log por pelo menos 7 dias, OU os change streams devem estar habilitados para que o DTS possa assinar alterações de dados dos últimos 7 dias. Se nenhuma dessas condições for atendida, o DTS poderá falhar ao buscar alterações no source, resultando em perda ou inconsistência de dados fora do escopo do service level agreement (SLA).

  Importante

  • Use o oplog para registrar alterações no banco de dados source (recomendado). - Os change streams exigem MongoDB 4.0 ou posterior. A sincronização bidirecional não é suportada ao usar change streams. - Para clusters não elásticos do Amazon DocumentDB, ative os change streams e defina Migration Method como ChangeStream e Architecture como Sharded Cluster.

• Durante a sincronização completa de dados, não modifique schemas de bancos de dados ou coleções, nem altere dados do tipo ARRAY.

• Para sources do tipo sharded cluster, não execute os seguintes comandos durante a sincronização: shardCollection, reshardCollection, unshardCollection, moveCollection ou movePrimary. Esses comandos podem causar inconsistência de dados.

• Se o banco de dados source for uma instância do MongoDB com arquitetura de sharded cluster e o balancer estiver balanceando dados, poderá ocorrer latência na instância.

Outros limites

• Apenas a sincronização no nível de coleção é suportada.

• Os bancos de dados admin, config e local não podem ser sincronizados.

• Um único registro superior a 10 MB causa falha na tarefa.

• As transações não são preservadas. O DTS converte cada transação em registros individuais no destino.

• Se nós broker forem adicionados ou removidos na instância do Kafka de destino enquanto uma tarefa do DTS estiver em execução, reinicie a tarefa do DTS.

• Certifique-se de que o DTS consiga se conectar às instâncias source e de destino. Por exemplo, as configurações de segurança das instâncias de banco de dados e os parâmetros listeners e advertised.listeners no arquivo server.properties de uma instância auto-gerenciada do Kafka não devem restringir o acesso do DTS.

• Execute a sincronização fora do horário de pico, quando a utilização da CPU nos bancos de dados source e de destino estiver abaixo de 30%.

• O DTS tenta automaticamente retomar instâncias com falha que estão em execução há menos de 7 days. Antes de alternar o tráfego para o destino, pare ou libere a instância de sincronização para evitar que a retomada automática sobrescreva os dados de destino.

• O DTS calcula a latência da sincronização incremental de dados com base no timestamp dos dados sincronizados mais recentes no banco de dados de destino e no timestamp atual no banco de dados source. Se nenhuma operação de atualização for realizada no banco de dados source por um longo período, a latência de sincronização poderá ficar imprecisa. Caso a latência da tarefa de sincronização esteja excessivamente alta, realize uma operação de atualização no banco de dados source para corrigi-la.

• Se uma tarefa do DTS falhar, o suporte técnico do DTS tentará restaurá-la dentro de 8 horas. Durante a restauração, a tarefa pode ser reiniciada e seus parâmetros podem ser modificados. Apenas os parâmetros da tarefa do DTS podem ser alterados; os parâmetros dos bancos de dados não são modificados. Os parâmetros passíveis de modificação incluem, mas não se limitam àqueles listados na seção "Modify instance parameters".

• Para sources do tipo sharded cluster com Oplog como método de sincronização incremental, o DTS não garante a ordem de gravação entre shards no tópico do Kafka de destino.

• Para sources do tipo sharded cluster, desative o balancer do MongoDB durante a sincronização completa de dados. Reative-o somente após a conclusão da sincronização completa e o início da sincronização incremental. Consulte Manage the ApsaraDB for MongoDB balancer.

Etapa 1: Abrir a página Data Synchronization

Utilize o console do DTS ou o console do DMS.

Console do DTS

1. Faça login no DTS console.DTS console

2. No painel de navegação à esquerda, clique em Data Synchronization.

3. No canto superior esquerdo, selecione a região onde a tarefa de sincronização está localizada.

Console do DMS

As etapas podem variar dependendo do modo do console do DMS. Consulte Simple mode e Customize the layout and style of the DMS console .

1. Faça login no DMS console.DMS console

2. Na barra de navegação superior, passe o ponteiro sobre Data + AI e escolha DTS (DTS) > Data Synchronization.

3. Na lista suspensa à direita de Data Synchronization Tasks, selecione a região.

Etapa 2: Configurar os bancos de dados source e de destino

Clique em Create Task e configure os parâmetros descritos nas tabelas a seguir.

Geral

Banco de dados source

Banco de dados de destino

Etapa 3: Testar a conectividade

Clique em Test Connectivity and Proceed na parte inferior da página.

O DTS adiciona seus blocos CIDR de servidor às configurações de segurança dos bancos de dados source e de destino automaticamente, se permitido. Para configuração manual, consulte Add the CIDR blocks of DTS servers .

Para bancos de dados auto-gerenciados que não utilizam Alibaba Cloud Instance como método de acesso, clique em Test Connectivity na caixa de diálogo CIDR Blocks of DTS Servers .

Etapa 4: Configurar objetos para sincronização

Na etapa Configure Objects, defina os seguintes parâmetros.

Etapa 5: Configurar definições avançadas

Clique em Next: Advanced Settings e configure os seguintes parâmetros.

Etapa 6: Executar a pré-verificação

Clique em Next: Save Task Settings and Precheck.

Para visualizar os parâmetros da API para esta configuração, passe o mouse sobre o botão e clique em Preview OpenAPI parameters antes de prosseguir.

A tarefa não pode ser iniciada até que a pré-verificação seja aprovada.

Se a pré-verificação falhar, clique em View Details ao lado de cada item com falha, resolva o problema e clique em Precheck Again .

Se um alerta for gerado para um item: corrija alertas não ignoráveis antes de prosseguir. Para alertas ignoráveis, clique em Confirm Alert Details , depois em Ignore na caixa de diálogo, confirme e clique em Precheck Again . Ignorar alertas pode causar inconsistência de dados.

Etapa 7: Adquirir e iniciar a instância

1. Aguarde até que a Success Rate atinja 100% e clique em Next: Purchase Instance.

2. Na página buy, configure os seguintes parâmetros.

   Parâmetro	Descrição
   Billing Method	Subscription: pagamento antecipado; mais econômico para uso de longo prazo. Pay-as-you-go: faturamento por hora; adequado para uso de curto prazo. Libere a instância quando não for mais necessária para interromper o faturamento.
   Resource Group Settings	Grupo de recursos para a instância de sincronização. Padrão: default resource group. Consulte What is Resource Management?
   Instance Class	Categoria de velocidade de sincronização. Consulte Instance classes of data synchronization instances.
   Subscription Duration	Disponível para faturamento por Subscription. Opções: 1–9 meses, 1 ano, 2 anos, 3 anos ou 5 anos.

3. Leia e aceite os Data Transmission Service (Pay-as-you-go) Service Terms.

4. Clique em Buy and Start e, em seguida, clique em OK na caixa de diálogo de confirmação.

A tarefa aparecerá na lista de tarefas. Acompanhe o progresso por lá.

Escolha seu cenário

Campos da mensagem Canal JSON

Todos os três cenários usam o mesmo envelope Canal JSON de nível superior. Os seguintes campos aparecem em todas as mensagens.

Cenário 1: Oplog

Defina o Migration Method como Oplog.

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

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

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

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

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

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

Cenário 2: ChangeStream — apenas campos atualizados

Defina o Migration Method como ChangeStream. Defina Obtain the entire document after it is updated como No.

As mensagens de inserção e exclusão são idênticas ao Cenário 1. As mensagens de atualização contêm apenas os campos alterados.

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

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

Cenário 3: ChangeStream — documento completo na atualização

Defina o Migration Method como ChangeStream. Defina Obtain the entire document after it is updated como Yes.

Os eventos de atualização entregam o documento completo após a alteração, em vez de apenas os campos modificados.

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

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

{
    "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: fullDocument ausente no ChangeStream

Quando o campo fullDocument de um evento de atualização do ChangeStream está ausente — por exemplo, quando um documento se move entre shards em uma coleção fragmentada — a mensagem entregue reverte para o comportamento do Oplog (atualização parcial com operadores $set ou $unset).

Exemplo: atualização de coleção fragmentada onde fullDocument está ausente

Dados base do source:

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

Alteração incremental no source:

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

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