Sincronizar dados do MySQL para o Elasticsearch em tempo real - Elasticsearch: Central de Documentação do Alibaba Cloud

Para pesquisar e analisar os dados de produção do seu banco de dados ApsaraDB RDS for MySQL com o Alibaba Cloud Elasticsearch, utilize o Data Transmission Service (DTS) para sincronizar os dados com um cluster Elasticsearch em tempo real. Essa sincronização ocorre por meio de uma tarefa de sincronização em tempo real, sendo ideal para cenários que exigem alto desempenho na transferência contínua de dados. Este tópico descreve como criar essa tarefa para mover dados de um banco de dados ApsaraDB RDS for MySQL para um cluster Alibaba Cloud Elasticsearch. Também explicamos como verificar os resultados da sincronização completa e incremental.

Informações básicas

O DTS é um serviço de transmissão de dados que integra migração, assinatura e sincronização de dados em tempo real. Para mais informações, consulte DTS. O serviço suporta a sincronização de alterações geradas por operações de inserção, exclusão e atualização. Para saber quais versões de fontes de dados são compatíveis, acesse Visão geral dos cenários de sincronização de dados.
É possível usar o DTS para sincronizar dados completos ou incrementais do MySQL para o Elasticsearch. Essa solução atende a casos que demandam alta performance na sincronização em tempo real a partir de bancos relacionais, ou quando há necessidade de transferir volumes totais ou parciais desses bancos para um cluster Alibaba Cloud Elasticsearch.

Precauções

O DTS não sincroniza alterações provenientes de operações DDL. Caso uma operação DDL seja executada em uma tabela do banco de dados de origem durante a sincronização, siga estes passos: remova a tabela da tarefa de sincronização, exclua o índice correspondente no cluster Elasticsearch e adicione a tabela novamente à tarefa. Para detalhes, veja Remover um objeto de sincronização e Adicionar um objeto a uma tarefa de sincronização de dados.
Ao adicionar colunas à tabela de origem, modifique os mapeamentos do índice associado a ela. Em seguida, execute a operação DDL correspondente na tabela de origem, pause a tarefa de sincronização e reinicie-a.
Durante a sincronização inicial completa, o DTS consome recursos de leitura e gravação tanto na origem quanto no destino, o que pode aumentar a carga nesses ambientes. Se o desempenho for baixo, as especificações forem limitadas ou o volume de dados for grande, a indisponibilidade pode ocorrer. Por exemplo, o consumo excessivo de recursos acontece quando há muitas consultas SQL lentas na origem, ausência de chaves primárias em tabelas ou deadlocks no destino. Para evitar problemas, avalie o impacto da sincronização antes de iniciá-la e prefira executar o processo em horários de baixa demanda, como quando a utilização da CPU na origem e no destino estiver abaixo de 30%.
- A sincronização de dados completos em horários de pico pode falhar. Nesse caso, reinicie a tarefa.
- Já a sincronização incremental nesse mesmo período pode apresentar latência elevada.
Como o ApsaraDB RDS for MySQL e o Elasticsearch suportam tipos de dados diferentes, o DTS estabelece mapeamentos entre os campos de origem e destino durante a sincronização inicial de esquema, baseando-se nos tipos aceitos pelo destino. Consulte Mapeamentos de tipos de dados para sincronização de esquema para mais detalhes.

Fluxo

Preparação: Adicione os dados a serem sincronizados ao banco de dados ApsaraDB RDS for MySQL de origem, crie um cluster Alibaba Cloud Elasticsearch e ative o recurso Auto Indexing no cluster.
Criação e execução da tarefa: Configure e inicie a tarefa de sincronização no console do DTS.
Verificação dos resultados: Acesse o console do Kibana do cluster Elasticsearch para validar a sincronização dos dados completos. Depois, insira novos registros no banco ApsaraDB RDS for MySQL e retorne ao Kibana para confirmar a replicação incremental.

Procedimento

Etapa 1: Preparação

Neste exemplo, utilizamos uma instância ApsaraDB RDS com MySQL 8.0 e um cluster Alibaba Cloud Elasticsearch V7.10.

Preparar o banco de dados de origem e os dados

Crie uma instância ApsaraDB RDS com MySQL 8.0. Para instruções detalhadas, consulte Criar uma instância ApsaraDB RDS for MySQL.
Configure uma conta e um banco de dados chamado test_mysql. Siga as orientações em Criar bancos de dados e contas.

No banco test_mysql, crie a tabela es_test e insira os dados conforme os comandos abaixo:

-- create table
CREATE TABLE `es_test` (
    `id` bigint(32) NOT NULL,
    `name` varchar(32) NULL,
    `age` bigint(32) NULL,
    `hobby` varchar(32) NULL,
    PRIMARY KEY (`id`)
) ENGINE=InnoDB
DEFAULT CHARACTER SET=utf8;

-- insert data
INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (1,'user1',22,'music');
INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (2,'user2',23,'sport');
INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (3,'user3',43,'game');
INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (4,'user4',24,'run');
INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (5,'user5',42,'basketball');

Preparar o cluster Elasticsearch de destino

Provisione um cluster Alibaba Cloud Elasticsearch V7.10. Veja o passo a passo em Criar um cluster Alibaba Cloud Elasticsearch.
Ative o recurso Auto Indexing no cluster. As instruções estão disponíveis em Configurar o arquivo YML.

Etapa 2: Criar e executar a tarefa de sincronização

Acesse a página Data Synchronization do novo console do DTS.
Clique em Create Task.
Na página exibida, crie e configure a tarefa seguindo as instruções na tela.

Nota
Para detalhes sobre os parâmetros das etapas seguintes, consulte Sincronizar dados de uma instância ApsaraDB RDS for MySQL para um cluster Elasticsearch.
1. Defina a origem e o destino. Na parte inferior da página, clique em Test Connectivity and Proceed.
2. Selecione os objetos que deseja sincronizar.
3. Ajuste as configurações avançadas. Neste exemplo, mantemos os valores padrão.
4. Na subetapa Data Verification, selecione Apply _routing Policy to No Tables.
  
  Nota
  Se o cluster Elasticsearch de destino for da versão V7.X, a opção Apply _routing Policy to No Tables é obrigatória.
Após concluir a configuração, salve a tarefa, execute a verificação prévia e adquira uma instância do DTS para iniciar a sincronização.

Com a aquisição da instância, a tarefa começa a rodar automaticamente. Acompanhe o progresso na página Data Synchronization. Assim que a carga completa terminar, os dados estarão disponíveis no cluster Elasticsearch.

Etapa 3 (Opcional): Validar os resultados da sincronização

Faça login no console do Kibana do cluster Elasticsearch.

Consulte Fazer login no console do Kibana para obter orientações.
No canto superior esquerdo do Kibana, navegue até > Management > Dev Tools e execute os comandos abaixo no Console.

Valide a sincronização dos dados completos.

Execute o seguinte comando:

GET /es_test/_search

Se a execução for bem-sucedida, o retorno será semelhante a este:

{
  "took" : 10,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 5,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "3",
        "_score" : 1.0,
        "_source" : {
          "id" : 3,
          "name" : "user3",
          "age" : 43,
          "hobby" : "game"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "5",
        "_score" : 1.0,
        "_source" : {
          "id" : 5,
          "name" : "user5",
          "age" : 42,
          "hobby" : "basketball"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "4",
        "_score" : 1.0,
        "_source" : {
          "id" : 4,
          "name" : "user4",
          "age" : 24,
          "hobby" : "run"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "id" : 2,
          "name" : "user2",
          "age" : 23,
          "hobby" : "sport"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "id" : 1,
          "name" : "user1",
          "age" : 22,
          "hobby" : "music"
        }
      }
    ]
  }
}

Confira a sincronização dos dados incrementais.

Insira um novo registro na tabela de origem com a instrução abaixo:

INSERT INTO `test_mysql`.`es_test` (`id`,`name`,`age`,`hobby`) VALUES (6,'user6',30,'dance');

Quando a sincronização incremental terminar, rode novamente o comando GET /es_test/_search para validar a atualização.

Uma execução bem-sucedida retorna o seguinte resultado:

{
  "took" : 541,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 6,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "3",
        "_score" : 1.0,
        "_source" : {
          "id" : 3,
          "name" : "user3",
          "age" : 43,
          "hobby" : "game"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "5",
        "_score" : 1.0,
        "_source" : {
          "id" : 5,
          "name" : "user5",
          "age" : 42,
          "hobby" : "basketball"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "4",
        "_score" : 1.0,
        "_source" : {
          "id" : 4,
          "name" : "user4",
          "age" : 24,
          "hobby" : "run"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "id" : 2,
          "name" : "user2",
          "age" : 23,
          "hobby" : "sport"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "6",
        "_score" : 1.0,
        "_source" : {
          "name" : "user6",
          "id" : 6,
          "age" : 30,
          "hobby" : "dance"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "id" : 1,
          "name" : "user1",
          "age" : 22,
          "hobby" : "music"
        }
      }
    ]
  }
}