Elasticsearch:Sincronizar dados do MySQL para o Alibaba Cloud Elasticsearch com o DataWorks

Etapa 1: Preparar os dados de origem no MySQL

Crie o banco de dados e a tabela na sua instância ApsaraDB RDS for MySQL.

1. Crie um banco de dados. Consulte Fluxo de trabalho geral para usar o ApsaraDB RDS for MySQL.

2. Crie uma tabela e insira dados:

   -- 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');

Etapa 2: Adquirir e configurar um grupo de recursos do Data Integration****

As tarefas de sincronização de dados são executadas em um grupo de recursos do Data Integration. Você deve adquirir um grupo e associar a rede e o workspace necessários.

1. Faça login no console do DataWorks.

2. Na barra de menu superior, selecione sua região. No painel de navegação à esquerda, clique em Resource Groups.

3. Clique em Create Resource Group, configure os parâmetros necessários e clique em Buy Now.

4. Associe uma VPC para conectividade de rede:

   1. Na coluna Actions do grupo de recursos recém-criado, clique em Network Settings.

   2. Na seção Data Scheduling & Data Integration, clique em Add VPC Association e associe a VPC à qual sua instância RDS e seu cluster Elasticsearch pertencem. Isso garante que o Data Integration possa se conectar a ambos.

   Importante

   Após associar uma VPC, adicione o bloco CIDR do vSwitch da VPC às listas de permissões internas dos clusters ApsaraDB RDS for MySQL e Elasticsearch. Para mais informações, consulte Configurar uma lista de permissões de endereços IP na documentação do ApsaraDB RDS for MySQL e Gerenciar listas de permissões de endereços IP na documentação do Alibaba Cloud Elasticsearch.

5. Associe o workspace do DataWorks:

   1. Retorne à página Resource Groups.

   2. Na coluna Actions do grupo de recursos, clique em Attach Workspace e selecione o workspace do DataWorks desejado.

Etapa 3: Adicionar fontes de dados no DataWorks

Adicione as fontes de dados ApsaraDB RDS for MySQL e ElasticSearch ao serviço Data Integration no DataWorks.

1. Acesse a página Data Integration no DataWorks.

   1. Faça login no console do DataWorks.

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

   3. Na coluna Actions do workspace desejado, escolha Shortcuts > Data Integration.

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

3. Adicione uma fonte de dados MySQL.

   1. Na página Data Sources, clique em Add Data Source.

   2. Na página Add Data Source, pesquise e selecione MySQL.

   3. Na caixa de diálogo Add MySQL Data Source, configure os parâmetros da fonte de dados na seção Basic Information.

      Para mais informações, consulte Configurar uma fonte de dados MySQL.

   4. Na seção Connection Configuration, clique em Test Connectivity. Se o status de conectividade for Connected, a conexão foi estabelecida com sucesso.

   5. Clique em Complete.

4. Adicione uma fonte de dados Elasticsearch. Consulte Adicionar uma fonte de dados Elasticsearch para obter instruções detalhadas.

Etapa 4: Configurar e executar uma tarefa de sincronização de dados em lote

Crie e execute uma tarefa de sincronização em lote (offline) no DataWorks. Este exemplo usa a interface visual sem código.

1. Acesse o Data Studio.

   1. Faça login no console do DataWorks.

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

   3. Na coluna Actions do workspace desejado, escolha Shortcuts > Data Studio.

2. Crie uma tarefa de sincronização.

   1. Na aba Data Studio (ícone ), escolha Create > Create Workflow e siga as instruções na tela para configurar o fluxo de trabalho.

   2. Adicione um nó ao fluxo de trabalho.

   3. Durante a criação do nó, defina o tipo de nó como Data Integration, escolha MySQL como origem e Elasticsearch como destino, configure-o como tipo em lote e insira um nome.

3. Configure a rede e os recursos.

   1. Na seção Data Source, defina Data Source como MySQL e selecione a fonte de dados adicionada anteriormente.

   2. Na seção My Resource Group, selecione o grupo de recursos do Data Integration adquirido.

   3. Na seção Data Destination, defina Data Destination como Elasticsearch e selecione o Data Source Name adicionado anteriormente.

4. Clique em Next.

5. Configure os detalhes da tarefa.

   1. Na seção Data Source, selecione a tabela específica do MySQL (por exemplo, es_test) a ser replicada.

   2. Na seção Data Destination, configure os parâmetros específicos do Elasticsearch (por exemplo, índice de destino, chave primária).

   3. Na seção Field Mapping, defina o mapeamento entre os campos de origem e de destino.

      Importante

      Certifique-se de validar explicitamente os mapeamentos de campos e alinhá-los corretamente pelo nome do campo, não pela posição ou ordem inferida automaticamente. O desalinhamento (por exemplo, mapear um campo de string de origem como name para um campo numérico de destino como id do tipo long) fará com que o Elasticsearch rejeite o documento, resultando em erros de "dados sujos".

   Para mais informações, consulte Configuração da interface visual sem código.

6. Execute a tarefa.

   1. (Opcional) Configure as propriedades de agendamento da tarefa. Na barra lateral direita, clique em Scheduling e configure os parâmetros. Para mais informações, consulte Agendamento.

   2. Salve a configuração da tarefa.

   3. Execute a tarefa.

      • Se o agendamento tiver sido configurado, a tarefa será executada automaticamente.

      • Para executar imediatamente, clique no ícone Run.

      Se o log contiver Shell run successfully!, a tarefa foi executada com sucesso.

Etapa 5: Verificar o resultado da sincronização de dados

1. Faça login no console do Kibana.

2. No console do Kibana, escolha Dev Tools.

3. No Console, execute o comando a seguir para visualizar os dados sincronizados.

   POST /es_test/_search?pretty
   {
   "query": { "match_all": {}}
   }

   Substitua es_test pelo nome do índice especificado na tarefa de sincronização de dados.

   Se os dados forem sincronizados com sucesso, o seguinte resultado será retornado.