Elasticsearch:Use DTS to synchronize data from PolarDB for MySQL to Alibaba Cloud Elasticsearch

Precautions

• DTS does not synchronize data changes generated by DDL operations. If a DDL operation is performed on a table in the source database during data synchronization, you must perform the following operations: Remove the table from the data synchronization task, remove the index for the table from the Elasticsearch cluster, and then add the table to the data synchronization task again. For more information, see Remove an object from a data synchronization task and Add an object to a data synchronization task.

• If you want to add columns to the source table, modify the mappings of the index that corresponds to the table. Then, perform the related DDL operation on the source table, pause the data synchronization task, and start the task again.

• Full data synchronization consumes read and write resources on both the source and destination databases, which can increase database load. If your database has poor performance, low specifications, or heavy workloads (for example, slow SQL queries, tables without primary keys, or destination deadlocks), the load increase may cause service unavailability. Evaluate database performance beforehand and perform data synchronization during off-peak hours when CPU utilization of both databases is below 30%.

  • If a full data synchronization task fails during peak hours, you can restart it.

  • If you perform incremental data synchronization during peak hours, data synchronization latency may occur.

Step 1: Make preparations

In this example, a PolarDB for MySQL 8.0.1 cluster of Enterprise Edition and an Alibaba Cloud Elasticsearch V7.10 cluster are prepared.

Step 2: Create and run a data synchronization task

1. Log on to the DTS data synchronization task list page (new version).

2. Click Create Task.

3. On the page that appears, create and configure a data synchronization task as prompted.

   For information about the parameters that are involved in the following steps, see Synchronize data from a PolarDB for MySQL cluster.

   1. Configure the source and destination databases, and then click Test Connectivity and Proceed.

      Source database information:

      • Set Database Type to PolarDB for MySQL

      • Set Access Method to Cloud Instance, and set Instance Region to China (Hangzhou)

      • Set Cross-account to No

      • Select the PolarDB Instance ID, and enter the Database Account and Database Password

      • Set Connection Type to Non-encrypted Connection

      Destination database information:

      • Set Database Type to Elasticsearch

      • Set Access Method to Cloud Instance, and set Instance Region to China (Hangzhou)

      • Set Type to Cluster

      • Select the Instance ID, set Database Account to elastic, and enter the Database Password

   2. Configure the objects from which you want to synchronize data.

      • Synchronization Type: Select Schema Synchronization, Full Data Synchronization, and Incremental Data Synchronization

      • Index Name: Select TableName

      • Processing Mode of Conflicting Tables: Select Pre-check and Report Errors

      • Case Sensitivity Policy for Object Names in the Destination Database: Select DTS Default Policy

      • In the Source Objects section, move the product table in the test_polardb database to the Selected Objects section

   3. Configure advanced settings. In this example, the default advanced settings are used.

   4. On the Database/Table/Column Settings page, click Set All to Non-_routing Policy.

      If the destination Elasticsearch cluster runs a 7.x version, you must set the routing policy to non-routing for all tables.

4. After the configuration is complete, save the data synchronization task, perform a pre-check on the task, and purchase a DTS instance to start the data synchronization task.

   After you purchase the DTS instance, the data synchronization task starts automatically. You can view progress on the Data Synchronization page. After full data synchronization completes, you can query the synchronized data in the Elasticsearch cluster.

   In the Running Phase Information column, you can view the synchronization progress, which includes three phases: Schema Synchronization, Full Data Synchronization, and Incremental Data Synchronization. A green check mark is displayed after each phase is complete.

Step 3 (Optional): Verify the data synchronization results

1. Log on to the Kibana console of the Elasticsearch cluster.

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

2. On the Kibana page, go to > Management > Dev Tools. In the Console, run the following commands.

3. Verify the synchronization result of full data.

   Run the following command:

   GET /product/_search

   If the command is successfully run, the following result is returned:

   {
     "took" : 3,
     "timed_out" : false,
     "_shards" : {
       "total" : 5,
       "successful" : 5,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : {
         "value" : 5,
         "relation" : "eq"
       },
       "max_score" : 1.0,
       "hits" : [
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "3",
           "_score" : 1.0,
           "_source" : {
             "id" : 3,
             "name" : "mobile phone C",
             "price" : "2600",
             "code" : "cmp",
             "color" : "black"
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "5",
           "_score" : 1.0,
           "_source" : {
             "id" : 5,
             "name" : "mobile phone E",
             "price" : "2800",
             "code" : "emp",
             "color" : "silvery"
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "4",
           "_score" : 1.0,
           "_source" : {
             "id" : 4,
             "name" : "mobile phone D",
             "price" : "2700",
             "code" : "dmp",
             "color" : "red"
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "2",
           "_score" : 1.0,
           "_source" : {
             "id" : 2,
             "name" : "mobile phone B",
             "price" : "2200",
             "code" : "bmp",
             "color" : "white"
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "1",
           "_score" : 1.0,
           "_source" : {
             "id" : 1,
             "name" : "mobile phone A",
             "price" : "2000",
             "code" : "amp",
             "color" : "golden"
           }
         }
       ]
     }
   }

4. Verify the synchronization result of incremental data.

   1. Execute the following statement to insert a data record into the source table:

      INSERT INTO `test_polardb`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (6,'mobile phone F','2750','fmp','white');

   2. After incremental data synchronization is complete, run the GET /product/_search command again to view the result.

      If the command is successfully run, the following result is returned:

      {
        "took" : 439,
        "timed_out" : false,
        "_shards" : {
          "total" : 5,
          "successful" : 5,
          "skipped" : 0,
          "failed" : 0
        },
        "hits" : {
          "total" : {
            "value" : 6,
            "relation" : "eq"
          },
          "max_score" : 1.0,
          "hits" : [
            {
              "_index" : "product",
              "_type" : "product",
              "_id" : "3",
              "_score" : 1.0,
              "_source" : {
                "id" : 3,
                "name" : "mobile phone C",
                "price" : "2600",
                "code" : "cmp",
                "color" : "black"
              }
            },
            {
              "_index" : "product",
              "_type" : "product",
              "_id" : "5",
              "_score" : 1.0,
              "_source" : {
                "id" : 5,
                "name" : "mobile phone E",
                "price" : "2800",
                "code" : "emp",
                "color" : "silvery"
              }
            },
            {
              "_index" : "product",
              "_type" : "product",
              "_id" : "4",
              "_score" : 1.0,
              "_source" : {
                "id" : 4,
                "name" : "mobile phone D",
                "price" : "2700",
                "code" : "dmp",
                "color" : "red"
              }
            },
            {
              "_index" : "product",
              "_type" : "product",
              "_id" : "2",
              "_score" : 1.0,
              "_source" : {
                "id" : 2,
                "name" : "mobile phone B",
                "price" : "2200",
                "code" : "bmp",
                "color" : "white"
              }
            },
            {
              "_index" : "product",
              "_type" : "product",
              "_id" : "6",
              "_score" : 1.0,
              "_source" : {
                "code" : "fmp",
                "color" : "white",
                "price" : "2750",
                "name" : "mobile phone F",
                "id" : 6
              }
            },
            {
              "_index" : "product",
              "_type" : "product",
              "_id" : "1",
              "_score" : 1.0,
              "_source" : {
                "id" : 1,
                "name" : "mobile phone A",
                "price" : "2000",
                "code" : "amp",
                "color" : "golden"
              }
            }
          ]
        }
      }