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

Contexto

O DataWorks é uma plataforma completa de desenvolvimento e governança de big data, construída sobre engines de big data. Ele integra recursos como desenvolvimento de dados, agendamento de tarefas e gerenciamento de dados. Use as tarefas de sincronização no DataWorks para transferir dados de diversas fontes para o Alibaba Cloud Elasticsearch de forma ágil.

• Fontes de dados suportadas incluem:

  • Bancos de dados da Alibaba Cloud (MySQL, PostgreSQL, SQL Server, MongoDB e HBase)

  • Alibaba Cloud PolarDB-X (atualizado a partir do DRDS)

  • Alibaba Cloud MaxCompute

  • Alibaba Cloud OSS

  • Alibaba Cloud Tablestore

  • Versões autogerenciadas de HDFS, Oracle, FTP, DB2 e outros tipos de banco de dados suportados

• Cenários de uso:

  • Sincronização offline de big data para o Alibaba Cloud Elasticsearch. É possível sincronizar um banco de dados inteiro ou todos os dados de uma tabela específica. Para mais informações, consulte Offline synchronization of an entire MySQL database to Elasticsearch.

  • Sincronização em tempo real de big data para o Alibaba Cloud Elasticsearch. Este método suporta sincronização completa e incremental. Para mais informações, consulte Real-time synchronization of an entire MySQL database to Elasticsearch.

Pré-requisitos

• Crie uma instância do RDS for MySQL. Para mais informações, consulte Create an RDS for MySQL instance. Este tópico utiliza o MySQL 5.7 como exemplo.

• Crie uma instância do Alibaba Cloud Elasticsearch e ative o recurso de criação automática de índices. Para mais informações, consulte Create an Alibaba Cloud Elasticsearch instance e Configure YML parameters.

• Crie um workspace do DataWorks. Para mais informações, consulte Create a workspace.

Faturamento

• Para detalhes sobre as taxas de uma instância do Alibaba Cloud Elasticsearch, consulte Elasticsearch billable items.

• Para informações sobre os custos de um grupo de recursos do Data Integration, consulte Resource group fees.

Etapa 1: Prepare os dados de source

Crie um banco de dados e uma tabela na instância do RDS for MySQL.

• É possível utilizar um banco de dados RDS da Alibaba Cloud ou um banco autogerenciado em servidor local. Este tópico usa um banco de dados RDS for MySQL como exemplo. Para mais informações, consulte Quick start.

• O código de exemplo abaixo demonstra como criar uma tabela e inserir 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: Adquira e configure um grupo de recursos

Adquira um grupo de recursos do Data Integration e associe uma VPC e um workspace a esse grupo.

1. Faça login no DataWorks console.

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

3. Clique em Create Resource Group e configure os parâmetros conforme solicitado.

4. Na coluna Actions do grupo de recursos criado, clique em Network Settings para associar uma VPC ao grupo de recursos. Para mais informações, consulte Configure network settings.

   Para permitir a sincronização de dados, o grupo de recursos precisa se conectar tanto à instância do RDS for MySQL quanto à instância do Elasticsearch. Portanto, associe o grupo de recursos à mesma Virtual Private Cloud (VPC), Zone e VSwitch das suas instâncias. Para encontrar as informações de VPC das suas instâncias, consulte Switch the VPC and vSwitch for an RDS for MySQL instance e View the basic information of an Elasticsearch instance.

   Importante

   Após associar a VPC, adicione o bloco CIDR do vSwitch CIDR Block à lista de permissões de IP tanto da instância do RDS for MySQL quanto da instância do Elasticsearch. Para mais informações, consulte Configure an IP whitelist for an RDS for MySQL instance e Configure a public or private IP whitelist for an Elasticsearch instance.

5. No canto superior esquerdo da página, clique no ícone de voltar para retornar à página Resource List.

6. Na coluna Actions do grupo de recursos criado, clique em Bind workspace para associar o workspace desejado ao grupo de recursos.

Etapa 3: Adicione fontes de dados

Adicione as fontes de dados do RDS for MySQL e do Elasticsearch ao service Data Integration no DataWorks.

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

   1. Faça login no DataWorks console.

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

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

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

3. Adicione uma fonte de dados do RDS for 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 Configure a MySQL data source.

   4. Na seção Connection Configuration, clique em Test Connectivity. O status Connected indica que a conexão foi estabelecida com sucesso.

   5. Clique em Complete.

4. Adicione uma fonte de dados do Elasticsearch seguindo o mesmo procedimento. Para mais informações, consulte Configure an Elasticsearch data source.

Etapa 4: Configure e execute uma tarefa de sincronização offline

Uma tarefa de sincronização offline é executada em um grupo de recursos, que atua como recurso computacional. O grupo de recursos recupera dados da fonte configurada e os grava no Elasticsearch.

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

   1. Faça login no DataWorks console.

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

   3. Na coluna Actions do workspace desejado, escolha Quick access > Data development.

2. Crie um nó de sincronização offline.

   1. Na aba Data Studio (ícone ) no painel de navegação à esquerda, escolha Create > New business process. Em seguida, crie um workflow conforme solicitado.

   2. Clique com o botão direito no workflow criado e escolha Create Node > Data Integration > Batch Synchronization.

   3. Na caixa de diálogo Create Node, insira um nome para o nó e clique em OK.

3. Configure a rede e os recursos.

   1. Na área Source, em Source, selecione MySQL e, em Data Source Name, selecione o nome da fonte de dados a ser sincronizada.

   2. Na seção Resource Group, selecione um grupo de recursos.

   3. Na seção Destination, em Destination, selecione Elasticsearch e, em Data Source Name, selecione o nome da fonte de dados a ser sincronizada.

4. Clique em Next step.

5. Configure a tarefa.

   1. Na seção Source, selecione a tabela que deseja sincronizar.

   2. Na seção Destination, configure os parâmetros do destino.

   3. Na seção Field Mapping, configure os mapeamentos entre Source Column e Target Column.

   4. Na seção Channel Control, configure os parâmetros do canal.

   Para mais detalhes sobre a configuração, consulte Configure a synchronization task by using the wizard.

6. Execute a tarefa.

   1. (Opcional) Configure as propriedades de agendamento da tarefa. No painel à direita, clique em Properties e defina os parâmetros de agendamento conforme necessário. Para mais informações sobre os parâmetros, consulte Configure scheduling properties.

   2. No canto superior esquerdo do editor de nós, clique no ícone Salvar para salve a tarefa.

   3. No canto superior esquerdo do editor de nós, clique no ícone Enviar para envie a tarefa.

      Se você configurou propriedades de agendamento, a tarefa será executada automaticamente conforme o cronograma definido. Também é possível clique em no ícone Executar no canto superior esquerdo do editor de nós para execute a tarefa imediatamente.

      A mensagem Shell run successfully! no log operacional indica que a tarefa foi executada com sucesso.

Etapa 5: Verifique o resultado

1. Faça login no console Kibana da instância de destino do Alibaba Cloud Elasticsearch. Para mais informações, consulte Log on to the Kibana console.

2. No canto superior esquerdo da página do Kibana, clique em no ícone e selecione Dev Tools.

3. No Console, execute o comando abaixo para visualize os dados sincronizados.

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

   Nota

   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.

   GET /es_test/_search
   
   {
     "took" : 0,
     "timed_out" : false,
     "_shards" : {
       "total" : 1,
       "successful" : 1,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : 5,
       "max_score" : 1.0,
       "hits" : [
         {
           "_index" : "es_test",
           "_type" : "es_test",
           "_id" : "2",
           "_score" : 1.0,
           "_source" : {
             "age" : 23,
             "hobby" : "sport",
             "id" : 2,
             "name" : "user2"
           }
         },
         {
           "_index" : "es_test",
           "_type" : "es_test",
           "_id" : "1",
           "_score" : 1.0,
           "_source" : {
             "age" : 22,
             "hobby" : "music",
             "id" : 1,
             "name" : "user1"
           }
         }
       ]
     }
   }