Elasticsearch:Utiliser DTS pour synchroniser les données de PolarDB for MySQL vers Alibaba Cloud Elasticsearch

Précautions

• DTS ne synchronise pas les modifications de données générées par les opérations DDL. Si une opération DDL est effectuée sur une table de la base de données source pendant la synchronisation des données, vous devez procéder comme suit : supprimez la table de la tâche de synchronisation de données, supprimez l'index de la table du cluster Elasticsearch, puis ajoutez à nouveau la table à la tâche de synchronisation de données. Pour plus d'informations, consultez Supprimer un objet d'une tâche de synchronisation de données et Ajouter un objet à une tâche de synchronisation de données.

• Si vous souhaitez ajouter des colonnes à la table source, modifiez les mappages de l'index correspondant à la table. Ensuite, effectuez l'opération DDL associée sur la table source, mettez en pause la tâche de synchronisation de données, puis redémarrez-la.

• La synchronisation complète des données consomme des ressources de lecture et d'écriture sur les bases de données source et de destination, ce qui peut augmenter leur charge. Si votre base de données présente de mauvaises performances, des spécifications faibles ou des charges de travail lourdes (par exemple, des requêtes SQL lentes, des tables sans clés primaires ou des interblocages au niveau de la destination), l'augmentation de la charge peut entraîner une indisponibilité du service. Évaluez les performances de la base de données au préalable et effectuez la synchronisation des données pendant les heures creuses, lorsque l'utilisation du CPU des deux bases de données est inférieure à 30 %.

  • Si une tâche de synchronisation complète des données échoue pendant les heures de pointe, vous pouvez la redémarrer.

  • Si vous effectuez une synchronisation incrémentielle des données pendant les heures de pointe, une latence de synchronisation des données peut survenir.

Étape 1 : Effectuer les préparatifs

Dans cet exemple, un cluster PolarDB for MySQL 8.0.1 Enterprise Edition et un cluster Alibaba Cloud Elasticsearch V7.10 sont préparés.

Étape 2 : Créer et exécuter une tâche de synchronisation de données

1. Connectez-vous à la page de liste des tâches de synchronisation de données DTS (nouvelle version).

2. Cliquez sur Create Task.

3. Sur la page qui s'affiche, créez et configurez une tâche de synchronisation de données selon les instructions.

   Pour obtenir des informations sur les paramètres impliqués dans les étapes suivantes, consultez Synchroniser les données d'un cluster PolarDB for MySQL.

   1. Configurez les bases de données source et de destination, puis cliquez sur Test Connectivity and Proceed.

      Informations sur la base de données source :

      • Définissez Database Type sur PolarDB for MySQL

      • Définissez Access Method sur Cloud Instance et définissez Instance Region sur China (Hangzhou)

      • Définissez Cross-account sur No

      • Sélectionnez l'PolarDB Instance ID et saisissez le Database Account et le Database Password

      • Définissez Connection Type sur Non-encrypted Connection

      Informations sur la base de données de destination :

      • Définissez Database Type sur Elasticsearch

      • Définissez Access Method sur Cloud Instance et définissez Instance Region sur China (Hangzhou)

      • Définissez Type sur Cluster

      • Sélectionnez l'Instance ID, définissez Database Account sur elastic et saisissez le Database Password

   2. Configurez les objets dont vous souhaitez synchroniser les données.

      • Synchronization Type : sélectionnez Schema Synchronization, Full Data Synchronization et Incremental Data Synchronization

      • Index Name : sélectionnez TableName

      • Processing Mode of Conflicting Tables : sélectionnez Pre-check and Report Errors

      • Case Sensitivity Policy for Object Names in the Destination Database : sélectionnez DTS Default Policy

      • Dans la section Source Objects, déplacez la table product de la base de données test_polardb vers la section Selected Objects

   3. Configurez les paramètres avancés. Dans cet exemple, les paramètres avancés par défaut sont utilisés.

   4. Sur la page Database/Table/Column Settings, cliquez sur Set All to Non-_routing Policy.

      Si le cluster Elasticsearch de destination exécute une version 7.x, vous devez définir la stratégie de routage sur non-routage pour toutes les tables.

4. Une fois la configuration terminée, enregistrez la tâche de synchronisation de données, effectuez une pré-vérification de la tâche et achetez une instance DTS pour démarrer la tâche de synchronisation de données.

   Après l'achat de l'instance DTS, la tâche de synchronisation de données démarre automatiquement. Vous pouvez consulter la progression sur la page Data Synchronization. Une fois la synchronisation complète des données terminée, vous pouvez interroger les données synchronisées dans le cluster Elasticsearch.

   Dans la colonne Running Phase Information, vous pouvez visualiser la progression de la synchronisation, qui comprend trois phases : Schema Synchronization, Full Data Synchronization et Incremental Data Synchronization. Une coche verte s'affiche après chaque phase terminée.

Étape 3 (facultative) : Vérifier les résultats de la synchronisation des données

1. Connectez-vous à la console Kibana du cluster Elasticsearch.

   Pour plus d'informations, consultez Se connecter à la console Kibana.

2. Sur la page Kibana, accédez à > Management > Dev Tools. Dans la Console, exécutez les commandes suivantes.

3. Vérifiez le résultat de la synchronisation des données complètes.

   Exécutez la commande suivante :

   GET /product/_search

   Si la commande s'exécute correctement, le résultat suivant est renvoyé :

   {
     "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. Vérifiez le résultat de la synchronisation des données incrémentielles.

   1. Exécutez l'instruction suivante pour insérer un enregistrement de données dans la table source :

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

   2. Une fois la synchronisation incrémentielle des données terminée, exécutez à nouveau la commande GET /product/_search pour afficher le résultat.

      Si la commande s'exécute correctement, le résultat suivant est renvoyé :

      {
        "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"
              }
            }
          ]
        }
      }