Elasticsearch:Sincronizar dados do MaxCompute para o Alibaba Cloud Elasticsearch usando o DataWorks

Etapa 1: Preparar os dados de origem

Crie uma tabela no MaxCompute e importe dados de teste. Para mais informações, consulte Create a table e Import data.

Este tópico utiliza o seguinte esquema de tabela e dados de amostra:

• Esquema da tabela

  A tabela contém 7 campos e 1 campo de partição. Os campos são definidos da seguinte forma:

  • create_time (string, chave primária)

  • category (string)

  • brand (string)

  • buyer_id (string)

  • trans_num (bigint)

  • trans_amount (double)

  • click_cnt (bigint)

  O campo de partição é pt (bigint).

• Dados de amostra

  A tabela de origem contém os seguintes campos:

  • create_time: data da transação, por exemplo, 2020/6/1

  • category: categoria do produto, por exemplo, Outerwear, Fresh Food, Electronics, Bathroom

  • brand: nome da marca, por exemplo, Brand A a Brand G

  • buyer_id: ID do comprador, por exemplo, user1 a user13

  • trans_num: número de transações

  • trans_amount: valor da transação

  • click_cnt: número de cliques

  • pt: campo de partição, com valor 1

Etapa 2: Adquirir e configurar um grupo de recursos exclusivo

Adquira um grupo de recursos exclusivo para Integração de Dados e associe-o a uma VPC e a um workspace. Um grupo de recursos exclusivo garante transmissão de dados rápida e estável.

1. Faça login no console do DataWorks .

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

3. Na aba Exclusive Resource Group, escolha Create Resource Group of Old Version > Resource Group for Data Integration.

4. Na página de compra DataWorks Exclusive Resources (Subscription), defina Exclusive Resource Type como Exclusive Resources for Data Integration, insira um nome para o grupo de recursos e clique em Buy Now.

   Para mais informações sobre a configuração, consulte Step 1: Purchase a resource group.

5. Localize o grupo de recursos exclusivo criado e clique em Network Settings na coluna Actions para associá-lo a uma VPC. Para mais informações, consulte Bind a VPC.

   Nota

   Este tópico utiliza um grupo de recursos exclusivo para Integração de Dados para sincronizar dados por meio de uma VPC. Para informações sobre como sincronizar dados pela internet, consulte Configure an allowlist.

   Para habilitar a sincronização de dados, conecte o grupo de recursos exclusivo à VPC onde reside o cluster do Alibaba Cloud Elasticsearch. Portanto, associe o grupo de recursos exclusivo à Virtual Private Cloud (VPC), Zone e VSwitch do seu cluster Alibaba Cloud Elasticsearch. Para visualizar essas informações, consulte View the basic information of an Elasticsearch cluster.

   Importante

   Após associar o grupo de recursos exclusivo à VPC, adicione o vSwitch CIDR Block à lista de permissões de IP privado da VPC do cluster Alibaba Cloud Elasticsearch. Para mais informações, consulte Configure a public or private IP address allowlist for an Elasticsearch cluster.

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

7. Encontre o grupo de recursos exclusivo criado e clique em Associate Workspace na coluna Actions para associá-lo ao workspace de destino.

   Para mais informações, consulte Step 2: Associate a workspace.

Etapa 3: Adicionar fontes de dados

Adicione o MaxCompute e o Alibaba Cloud Elasticsearch como fontes de dados na Integração de Dados.

1. Acesse a página Data Integration.

   1. Faça login no console do DataWorks .

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

   3. Localize o workspace desejado e escolha Shortcuts > Data Integration na coluna Actions.

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

3. Adicione uma fonte de dados MaxCompute.

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

   2. Na página Add Data Source, localize e selecione o tipo de fonte de dados MaxCompute.

   3. Na caixa de diálogo Add MaxCompute data source, configure os parâmetros na seção Basic Information.

      Para mais informações, consulte Configure a MaxCompute data source.

   4. Na seção Connection Configuration, clique em Test Connectivity. O status Connected indica que a conexão foi bem-sucedida.

   5. Clique em Complete.

4. Siga o mesmo procedimento para adicionar uma fonte de dados Elasticsearch. Para mais informações, consulte Configure an Elasticsearch data source.

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

Uma tarefa de sincronização em lote usa o grupo de recursos exclusivo. O grupo de recursos recupera dados da fonte de dados na Integração de Dados e grava os dados no cluster do Alibaba Cloud Elasticsearch.

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

   1. Faça login no console do DataWorks .

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

   3. Localize o workspace desejado e escolha Shortcuts > Data development na coluna Actions.

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

   1. Na aba Data Development (), escolha Create > New business process.

   2. Clique com o botão direito no fluxo de trabalho 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 o grupo de recursos.

   1. Na seção Source, defina Source como MaxCompute(ODPS) e selecione sua fonte de dados de origem em Data Source Name.

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

   3. Na seção Destination, defina Destination como Elasticsearch e selecione sua fonte de dados de destino em Data Source Name.

4. Clique em Next step.

5. Configure a tarefa.

   1. Na seção Source, selecione a tabela de origem.

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

   3. Na seção Field Mapping, mapeie a Source Column para a Target Column.

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

   Para mais informações sobre a configuração, consulte Configure a synchronization task in wizard mode.

6. Execute a tarefa.

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

   2. No canto superior direito da área do nó, clique no ícone Salvar para salvar a tarefa.

   3. No canto superior direito da área do nó, clique no ícone Commit.

      Se você configurar propriedades de agendamento para a tarefa, ela será executada automaticamente em intervalos programados. Também é possível clicar no ícone Run no canto superior direito da área do nó para executar a tarefa imediatamente.

      A mensagem Shell run successfully! no log de execução indica que a tarefa foi concluída com êxito. Um exemplo do log de execução da tarefa é apresentado abaixo:

      2023-10-31 16:52:35 INFO Exit code of the Shell command 0
      2023-10-31 16:52:35 INFO --- Invocation of Shell command completed ---
      2023-10-31 16:52:35 INFO Shell run successfully!
      2023-10-31 16:52:35 INFO Current task status: FINISH
      2023-10-31 16:52:35 INFO Cost time is: 33.106s

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

No console do Kibana, visualize os dados sincronizados e execute consultas.

1. Faça login no console do Kibana do cluster Alibaba Cloud Elasticsearch de destino.

   Para mais informações, consulte Log on to the Kibana console.

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

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

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

   Nota

   odps_index é o valor do campo index definido no script de sincronização de dados.

   Uma sincronização bem-sucedida retorna o seguinte resultado.

   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {}}
   }
   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {} },
   "_source": ["category", "brand"]
   }
   POST /odps_index/_search?pretty
   {
     "query": { "match": {"category":"生鲜"} }
   }
   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {} },
   "sort": { "trans_num": { "order": "desc" }
           }
   }
   
   --- Response ---
   {
     "took" : 2,
     "timed_out" : false,
     "_shards" : {
       "total" : 1,
       "successful" : 1,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : 13,
       "max_score" : null,
       "hits" : [
         {
           "_index" : "odps_index",
           "_type" : "_doc",
           "_id" : "2020/6/7 8:00",
           "_score" : null,
           "_source" : {
             "trans_num" : 88,
             "click_cnt" : 80,
             "category" : "外套",
             "buyer_id" : "user7",
             "trans_amount" : 150.0,
             "brand" : "品牌E"
           },
           "sort" : [
             88
           ]
         },
         {
           "_index" : "odps_index",
           "_type" : "_doc",
           "_id" : "2020/6/11 8:00",
           "_score" : null,
           "_source" : {
             "trans_num" : 22,
             "click_cnt" : 70,
             "category" : "卫浴",
             "buyer_id" : "user11",
             "trans_amount" : 4500.0,
             "brand" : "品牌G"
           },
           "sort" : [
             22
           ]
         }
       ]
     }
   }

4. Execute o seguinte comando para buscar os campos category e brand no documento.

   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {} },
   "_source": ["category", "brand"]
   }

5. Execute o seguinte comando para pesquisar documentos onde category seja fresh food.

   POST /odps_index/_search?pretty
   {
   "query": { "match": {"category":"生鲜"} }
   }

6. Execute o seguinte comando para ordenar os documentos pelo campo trans_num.

   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {} },
   "sort": { "trans_num": { "order": "desc" } }
   }

   Para mais comandos e métodos de acesso, consulte o Centro de Ajuda Elastic.co.