Elasticsearch:Synchronize MaxCompute data to Alibaba Cloud Elasticsearch using DataWorks

Step 1: Prepare the source data

Create a MaxCompute table and import test data into the table. For more information, see Create a table and Import data.

This topic uses the following table schema and sample data:

• Table schema

  The table contains 7 fields and 1 partition field. The fields are defined as follows:

  • create_time (string, primary key)

  • category (string)

  • brand (string)

  • buyer_id (string)

  • trans_num (bigint)

  • trans_amount (double)

  • click_cnt (bigint)

  The partition field is pt (bigint).

• Sample data

  The source table contains the following fields:

  • create_time: transaction date, for example, 2020/6/1

  • category: product category, for example, Outerwear, Fresh Food, Electronics, Bathroom

  • brand: brand name, for example, Brand A to Brand G

  • buyer_id: buyer ID, for example, user1 to user13

  • trans_num: number of transactions

  • trans_amount: transaction amount

  • click_cnt: number of clicks

  • pt: partition field, with a value of 1

Step 2: Purchase and configure an exclusive resource group

Purchase an exclusive resource group for Data Integration and associate it with a VPC and a workspace. An exclusive resource group ensures fast and stable data transmission.

1. Log on to the DataWorks console .

2. In the top navigation bar, select a region. In the left-side navigation pane, click Resource Groups.

3. On the Exclusive Resource Group tab, choose Create Resource Group of Old Version > Resource Group for Data Integration.

4. On the DataWorks Exclusive Resources (Subscription) purchase page, set Exclusive Resource Type to Exclusive Resources for Data Integration, enter a name for the resource group, and click Buy Now.

   For more information about the configuration, see Step 1: Purchase a resource group.

5. Find the exclusive resource group you created and click Network Settings in the Actions column to associate it with a VPC. For more information, see Bind a VPC.

   Note

   This topic uses an exclusive resource group for Data Integration to synchronize data over a VPC. For information about how to synchronize data over the internet, see Configure an allowlist.

   To enable data synchronization, the exclusive resource group must be connected to the VPC where the Alibaba Cloud Elasticsearch cluster resides. Therefore, you must associate the exclusive resource group with the Virtual Private Cloud (VPC), Zone, and VSwitch of your Alibaba Cloud Elasticsearch cluster. To view this information, see View the basic information of an Elasticsearch cluster.

   Important

   After associating the exclusive resource group with the VPC, you must add the vSwitch CIDR Block to the VPC private IP allowlist of the Alibaba Cloud Elasticsearch cluster. For more information, see Configure a public or private IP address allowlist for an Elasticsearch cluster.

6. In the upper-left corner of the page, click the back icon to return to the Resource List page.

7. Find the exclusive resource group you created and click Associate Workspace in the Actions column to associate it with the target workspace.

   For more information, see Step 2: Associate a workspace.

Step 3: Add data sources

Add MaxCompute and Alibaba Cloud Elasticsearch as data sources in Data Integration.

1. Go to the Data Integration page.

   1. Log on to the DataWorks console .

   2. In the left-side navigation pane, click Workspace.

   3. Find the target workspace and choose Shortcuts > Data Integration in the Actions column.

2. In the left-side navigation pane, click Data Source.

3. Add a MaxCompute data source.

   1. On the Data Sources page, click Add Data Source.

   2. On the Add Data Source page, find and select the MaxCompute data source type.

   3. In the Add MaxCompute data source dialog box, configure the parameters in the Basic Information section.

      For more information, see Configure a MaxCompute data source.

   4. In the Connection Configuration section, click Test Connectivity. A status of Connected indicates that the connection is successful.

   5. Click Complete.

4. Follow the same procedure to add an Elasticsearch data source. For more information, see Configure an Elasticsearch data source.

Step 4: Configure and run a batch synchronization task

A batch synchronization task runs using the exclusive resource group. The resource group retrieves data from the data source in Data Integration and writes the data to the Alibaba Cloud Elasticsearch cluster.

1. Go to the Data development page in DataWorks.

   1. Log on to the DataWorks console .

   2. In the left-side navigation pane, click Workspace.

   3. Find the target workspace and choose Shortcuts > Data development in the Actions column.

2. Create an offline synchronization node.

   1. On the Data Development () tab, choose Create > New business process.

   2. Right-click the workflow that you created and choose Create Node > Data Integration > Batch Synchronization.

   3. In the Create Node dialog box, enter a name for the node and click OK.

3. Configure the network and resource group.

   1. In the Source section, set Source to MaxCompute(ODPS) and select your source data source under Data Source Name.

   2. In the Resource Group section, select the exclusive resource group.

   3. In the Destination section, set Destination to Elasticsearch and select your destination data source under Data Source Name.

4. Click Next step.

5. Configure the task.

   1. In the Source section, select the source table.

   2. In the Destination section, configure the parameters for the destination.

   3. In the Field Mapping section, map the Source Column to the Target Column.

   4. In the Channel Control section, configure the channel parameters.

   For more information about the configuration, see Configure a synchronization task in wizard mode.

6. Run the task.

   1. (Optional) Configure scheduling properties for the task. In the right-side pane, click Properties and configure the scheduling parameters as needed. For more information about the parameters, see Scheduling configuration.

   2. In the upper-right corner of the node area, click the Save icon to save the task.

   3. In the upper-right corner of the node area, click the Commit icon.

      If you configure scheduling properties for the task, the task runs automatically at scheduled intervals. You can also click the Run icon in the upper-right corner of the node area to run the task immediately.

      The message Shell run successfully! in the run log indicates that the task ran successfully. A sample of the task run log is as follows:

      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

Step 5: Verify the data synchronization result

In the Kibana console, view the synchronized data and run queries.

1. Log on to the Kibana console of the target Alibaba Cloud Elasticsearch cluster.

   For more information, see Log on to the Kibana console.

2. In the upper-left corner of the Kibana page, click the icon and select Dev Tools.

3. In the Console, run the following command to view the synchronized data.

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

   Note

   odps_index is the value of the index field that you set in the data synchronization script.

   A successful synchronization returns the following result.

   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. Run the following command to search for the category and brand fields in the document.

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

5. Run the following command to search for documents where category is fresh food.

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

6. Run the following command to sort the documents by the trans_num field.

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

   For more commands and access methods, see the Elastic.co Help Center.