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

Para pesquisar e analisar 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. A sincronização ocorre por meio de uma tarefa de sincronização em tempo real, sendo ideal para cenários que exigem alto desempenho. Este tópico descreve como criar essa tarefa para transferir dados de um banco ApsaraDB RDS for MySQL para um cluster Alibaba Cloud Elasticsearch, além de explicar como verificar os resultados da sincronização completa e incremental.

Informações básicas

O DTS é um service de transmissão de dados que integra migração, assinatura e sincronização em tempo real. Para mais detalhes, consulte DTS. Ele suporta a sincronização de alterações geradas por operações de inserção, exclusão e atualização. Para verificar as versões das fontes de dados compatíveis, veja Overview of data synchronization scenarios.
É possível usar o DTS para sincronizar dados completos ou incrementais do MySQL para o Elasticsearch. Essa solução atende a cenários que demandam alta performance na sincronização em tempo real a partir de bancos relacionais, ou quando há necessidade de transferir dados totais ou parciais para um cluster Alibaba Cloud Elasticsearch.

Precauções

O DTS não sincroniza alterações de dados geradas por operações DDL. Caso uma operação DDL seja executada em uma tabela do banco de dados source 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. Consulte Remove a synchronization object e Add an object to a data synchronization task para mais informações.
Ao adicionar colunas à tabela source, modifique os mapeamentos do índice associado a ela. Em seguida, execute a operação DDL correspondente na tabela source, pause a tarefa de sincronização e reinicie-a.
Durante a sincronização inicial completa, o DTS consome recursos de leitura e gravação da origem e do destino, o que pode aumentar a carga nesses sistemas. 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 no desempenho antes de iniciar o processo. Recomendamos executar a sincronização 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 nesses mesmos períodos pode apresentar latência.
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 suportados pelo destino. Veja mais em Data type mappings for schema synchronization.

Processo

Preparação: Adicione os dados a serem sincronizados ao banco de dados source ApsaraDB RDS for MySQL, 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 source ApsaraDB RDS for MySQL e retorne ao Kibana para confirmar a sincronizaçã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 source e os dados

Crie uma instância ApsaraDB RDS com MySQL 8.0. Para mais detalhes, consulte Create an ApsaraDB RDS for MySQL instance.
Crie uma conta de banco de dados e um banco chamado test_mysql. Veja mais em Create databases and accounts.

No banco test_mysql, crie a tabela es_test e insira dados. Utilize o seguinte comando de criação e inserção:

-- 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

Crie um cluster Alibaba Cloud Elasticsearch V7.10. Consulte Create an Alibaba Cloud Elasticsearch cluster para instruções detalhadas.
Ative o recurso Auto Indexing no cluster Elasticsearch. Mais informações em Configure the YML file.

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 conforme as instruções.

Nota
Para detalhes sobre os parâmetros das etapas seguintes, consulte Synchronize data from an ApsaraDB RDS for MySQL instance to an Elasticsearch cluster.
1. Configure a origem e o destino. Na parte inferior da página, clique em Test Connectivity and Proceed.
2. Defina os objetos cujos dados serão sincronizados.
3. Ajuste as configurações avançadas. Neste exemplo, mantenha os padrões.
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, é obrigatório selecionar Apply _routing Policy to No Tables.
Após concluir a configuração, salve a tarefa, execute a verificação prévia e adquira uma instância do DTS para iniciá-la.

Com a aquisição da instância, a sincronização começa automaticamente. Acompanhe o progresso na página Data Synchronization. Ao término da fase completa, os dados estarão disponíveis no cluster Elasticsearch.

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

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

Consulte Log on to the Kibana console para orientações.
No canto superior esquerdo do Kibana, selecione > 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"
        }
      }
    ]
  }
}

Confirme a sincronização dos dados incrementais.

Execute a instrução abaixo para inserir um registro na tabela source:

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

Após a conclusão da sincronização incremental, rode novamente o comando GET /es_test/_search para validar o resultado.

Em caso de sucesso, a resposta esperada é:

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