How to migrate data from Amazon DynamoDB to Alibaba Cloud using NimoShake - ApsaraDB for MongoDB: Central de Documentação do Alibaba Cloud

NimoShake (também conhecido como DynamoShake) é uma ferramenta de sincronização de dados desenvolvida pelo Alibaba Cloud para migrar bancos de dados Amazon DynamoDB para o ApsaraDB for MongoDB. Suporta migração completa, migração incremental ou ambas em uma única execução.

Suporte a recursos

Recurso	Suportado
Migração completa	Sim
Migração incremental	Sim
Transmissão retomável (incremental)	Sim
Transmissão retomável (completa)	Não
Migração de índices (somente fase completa)	Sim
Migração somente de schema	Sim
Filtragem de collections	Sim
Migração de índices (fase incremental)	Não

Pré-requisitos

Antes de começar, certifique-se de ter:

Uma instância de replica set ou de cluster sharded do ApsaraDB for MongoDB. Consulte Create a replica set instance ou Create a sharded cluster instance.
O AccessKey ID e o AccessKey secret do Amazon DynamoDB.
Espaço de armazenamento suficiente no ApsaraDB for MongoDB para todos os dados do banco de dados DynamoDB de origem. A capacidade de armazenamento do destino deve ser superior à da origem.

Como funciona

O NimoShake executa a migração completa e a migração incremental como fases separadas.

Migração completa

A migração completa é composta por duas partes: migração de dados seguida de migração de índices.

Full migration architecture

A migração de dados utiliza três tipos de threads em um pipeline:

Data migration threads

Thread	Descrição
Fetcher	Chama o driver de conversão de protocolo da Amazon para recuperar dados em lote da tabela de origem e os coloca em filas. Apenas um thread fetcher é executado por vez.
Parser	Lê os dados das filas, converte para o formato BSON e repassa aos executors. Padrão: 2 threads. Controlado por `full.document.parser`.
Executor	Extrai dados das filas, agrega até 16 MB ou 1.024 entradas e grava no destino. Padrão: 4 threads. Controlado por `full.document.concurrency`.

A migração de índices ocorre após a conclusão da migração de dados e cria os seguintes índices:

Índices gerados automaticamente:
- Se a tabela de origem tem uma partition key e uma sort key: um índice composto único em ambas as chaves, mais um índice hashed na partition key.
- Se a tabela de origem tem apenas uma partition key: um índice hashed e um índice único na partition key.
Índices criados pelo usuário: Um índice hashed baseado na chave primária é criado para cada índice definido pelo usuário.

Migração incremental

A migração incremental captura as alterações contínuas da origem e as grava no ApsaraDB for MongoDB. Índices criados durante a fase incremental não são migrados.

Incremental migration architecture

Thread	Descrição
Fetcher	Monitora as alterações de shard no stream.
Manager	Gerencia as notificações de mensagens e cria um Dispatcher para cada shard.
Dispatcher	Recupera dados incrementais da origem, retomando a partir do último checkpoint quando a transmissão retomável está ativa.
Batcher	Analisa, empacota e agrega dados incrementais provenientes dos threads Dispatcher.
Executor	Grava os dados agregados na instância ApsaraDB for MongoDB de destino e atualiza o checkpoint.

Transmissão retomável e checkpoints

A migração incremental suporta transmissão retomável por meio de checkpoints. Se a conexão for perdida e restaurada rapidamente, a migração retoma a partir do último checkpoint. Uma desconexão prolongada ou a perda do checkpoint pode acionar uma nova migração completa.

Por padrão, os checkpoints são armazenados no banco de dados ApsaraDB for MongoDB de destino, em um banco de dados chamado nimo-shake-checkpoint. Cada collection tem sua própria tabela de checkpoint, e uma tabela status_table registra se a sincronização atual é uma tarefa completa ou incremental.

A migração completa não suporta transmissão retomável. Se interrompida, reinicia do início.

Migrar do DynamoDB para o ApsaraDB for MongoDB

Os passos a seguir utilizam o Ubuntu como exemplo.

Etapa 1: Baixe o NimoShake

wget https://github.com/alibaba/NimoShake/releases/download/release-v1.0.14-20250704/nimo-shake-v1.0.14.tar.gz

Baixe a versão mais recente na página de releases do NimoShake .

Etapa 2: Extraia o pacote

tar zxvf nimo-shake-v1.0.14.tar.gz

Etapa 3: Acesse o diretório

cd nimo-shake-v1.0.14

Etapa 4: Configure o NimoShake

Abra o arquivo de configuração:

vi nimo-shake.conf

As tabelas abaixo descrevem todos os parâmetros de configuração, agrupados por categoria. Comece pelos parâmetros obrigatórios e ajuste os opcionais conforme necessário.

Parâmetros obrigatórios

Parâmetro	Descrição	Exemplo
`source.access_key_id`	O AccessKey ID do banco de dados Amazon DynamoDB.	`source.access_key_id = AKIAIOSFODNN7EXAMPLE`
`source.secret_access_key`	O AccessKey secret do banco de dados Amazon DynamoDB.	`source.secret_access_key = wJalrXUtnFEMI/K7MDENG`
`source.region`	A região AWS do banco de dados DynamoDB. Opcional se a região for detectada automaticamente ou não se aplicar.	`source.region = us-east-2`
`target.type`	O tipo do banco de dados de destino. Defina como `mongodb` para o ApsaraDB for MongoDB. Defina como `aliyun_dynamo_proxy` para uma instância ApsaraDB for MongoDB compatível com DynamoDB. Para mais endereços MongoDB, consulte Replica set instance connection ou Connect to a sharded cluster instance.	`target.type = mongodb`
`target.address`	A string de conexão do banco de dados de destino. Consulte Connect to a replica set instance ou Connect to a sharded cluster instance.	`target.address = mongodb://username:password@s-*****-pub.mongodb.rds.aliyuncs.com:3717`
`target.mongodb.type`	O tipo da instância ApsaraDB for MongoDB de destino. `replica` para uma instância de replica set. `sharding` para uma instância de cluster sharded.	`target.mongodb.type = sharding`
`sync_mode`	O modo de migração. `all`: executa a migração completa seguida de migração incremental. `full`: executa somente a migração completa. Padrão: `all`. Nota Somente `full` é suportado quando a origem é uma instância ApsaraDB for MongoDB compatível com DynamoDB.	`sync_mode = all`

Configurações gerais

Parâmetro	Padrão	Obrigatório	Descrição	Exemplo
`id`	—	Opcional	O ID da tarefa de migração. Usado para arquivos PID, nomes de log, o nome do banco de dados de checkpoint e o nome do banco de dados de destino.	`id = nimo-shake`
`log.file`	stdout	Opcional	O caminho do arquivo de log. Se não definido, os logs são impressos no stdout.	`log.file = nimo-shake.log`
`log.level`	`info`	Opcional	O nível de log. Valores válidos: `none`, `error`, `warn`, `info`, `debug`.	`log.level = info`
`log.buffer`	`true`	Opcional	Se o buffering de log deve ser ativado. `true`: alto desempenho, mas pode perder as últimas entradas de log ao encerrar. `false`: todas as entradas de log são gravadas, mas o desempenho pode ser reduzido.	`log.buffer = true`
`system_profile`	—	Opcional	A porta PPROF para depuração e visualização de informações de corrotinas com pilha.	`system_profile = 9330`
`full_sync.http_port`	—	Opcional	A porta RESTful para a fase de migração completa. Use `curl` para visualize estatísticas de monitoramento. Consulte o wiki.	`full_sync.http_port = 9341`
`incr_sync.http_port`	—	Opcional	A porta RESTful para a fase de migração incremental. Use `curl` para visualize estatísticas de monitoramento. Consulte o wiki.	`incr_sync.http_port = 9340`

Configurações de conexão com a origem

Parâmetro	Padrão	Obrigatório	Descrição	Exemplo
`source.session_token`	—	Opcional	O token de sessão temporário para acesso ao DynamoDB. Necessário somente ao usar credenciais temporárias.	`source.session_token = AQoXnyc4lcK4w4...`
`source.endpoint_url`	—	Opcional	A URL do endpoint, caso a origem seja do tipo endpoint. *Defina este parâmetro substitui todos os demais parâmetros `source.`.**	`source.endpoint_url = "http://192.168.0.1:1010"`
`source.session.max_retries`	—	Opcional	O número máximo de tentativas após uma falha de sessão.	`source.session.max_retries = 3`
`source.session.timeout`	—	Opcional	O timeout da sessão em milissegundos. Defina como `0` para desativar o timeout.	`source.session.timeout = 3000`

Filtragem de collections

Parâmetro	Padrão	Obrigatório	Descrição	Exemplo
`filter.collection.white`	—	Opcional	Lista de permissões de collections a migrar. Somente as collections listadas são migradas.	`filter.collection.white = c1;c2`
`filter.collection.black`	—	Opcional	Lista de bloqueios de collections a excluir. Todas as demais collections são migradas. Não pode ser usado em conjunto com `filter.collection.white`. Se ambos forem definidos, todas as collections são migradas.	`filter.collection.black = c1;c2`

Configurações do destino

Parâmetro	Padrão	Obrigatório	Descrição	Exemplo
`target.db.exist`	Error	Opcional	Como tratar collections com o mesmo nome já existentes no destino. `rename`: renomeia a collection existente adicionando um sufixo de timestamp (por exemplo, `c1` torna-se `c1.2019-07-01Z12:10:11`). `drop`: exclui a collection existente. Se não definido, a migração é interrompida com erro caso exista uma collection com o mesmo nome.	`target.db.exist = drop`
`sync_schema_only`	`false`	Opcional	Se deve migrar apenas o schema da tabela, sem os dados.	`sync_schema_only = false`

Desempenho da migração completa

Parâmetro	Padrão	Obrigatório	Descrição	Exemplo
`full.concurrency`	`4`	Opcional	O número máximo de collections migradas simultaneamente.	`full.concurrency = 4`
`full.read.concurrency`	`1`	Opcional	O número de threads simultâneos de leitura de uma única tabela de origem. Corresponde ao parâmetro `TotalSegments` da API Scan do DynamoDB.	`full.read.concurrency = 1`
`full.document.concurrency`	`4`	Opcional	O número de threads de gravação simultâneos por tabela.	`full.document.concurrency = 4`
`full.document.write.batch`	—	Opcional	O número de entradas agregadas por gravação. Quando o destino é um banco de dados compatível com DynamoDB, o valor máximo é 25.	`full.document.write.batch = 25`
`full.document.parser`	`2`	Opcional	O número de threads de parser simultâneos para converter dados do DynamoDB para o protocolo de destino.	`full.document.parser = 2`
`full.enable_index.user`	`true`	Opcional	Se deve migrar os índices definidos pelo usuário.	`full.enable_index.user = true`
`full.executor.insert_on_dup_update`	`true`	Opcional	Se deve converter um `INSERT` em `UPDATE` quando já existe uma chave duplicada no destino.	`full.executor.insert_on_dup_update = true`
`qps.full`	`1000`	Opcional	O número máximo de chamadas do comando `Scan` por segundo durante a migração completa.	`qps.full = 1000`
`qps.full.batch_num`	`128`	Opcional	O número de entradas de dados recuperadas por segundo durante a migração completa.	`qps.full.batch_num = 128`
`full.read.filter_expression`	—	Opcional	Uma expressão de filtro do DynamoDB para a migração completa. As variáveis começam com dois pontos, por exemplo `:begin` e `:end`. Especifique os valores das variáveis em `full.read.filter_attributevalues`.	`full.read.filter_expression = create_time > :begin AND create_time < :end`
`full.read.filter_attributevalues`	—	Opcional	Os valores para as variáveis em `full.read.filter_expression`. `N` representa Number; `S` representa String.	`full.read.filter_attributevalues = beginN1646724207280~~~endN1646724207283`

Desempenho da migração incremental

Ignore esta seção se estiver executando apenas migração completa ( sync_mode = full ).

Parâmetro	Padrão	Obrigatório	Descrição	Exemplo
`incr_sync_parallel`	`false`	Opcional	Se deve ativar a migração incremental paralela. `true`: usa mais memória. `false`: modo padrão.	`incr_sync_parallel = false`
`increase.concurrency`	`16`	Opcional	O número máximo de shards capturados simultaneamente.	`increase.concurrency = 16`
`increase.executor.insert_on_dup_update`	`true`	Opcional	Se deve converter um `INSERT` em `UPDATE` quando as mesmas chaves já existem no destino.	`increase.executor.insert_on_dup_update = true`
`increase.executor.upsert`	`true`	Opcional	Se deve converter um `UPDATE` em `UPSERT` quando não existem chaves correspondentes no destino. Um `UPSERT` atualiza o registro se a chave existir ou o insere caso contrário.	`increase.executor.upsert = true`
`qps.incr`	`1000`	Opcional	O número máximo de chamadas do comando `GetRecords` por segundo durante a migração incremental.	`qps.incr = 1000`
`qps.incr.batch_num`	`128`	Opcional	O número de entradas de dados recuperadas por segundo durante a migração incremental.	`qps.incr.batch_num = 128`

Configurações de checkpoint