Elasticsearch:Synchroniser des données MySQL vers Alibaba Cloud Elasticsearch avec DataWorks

Contexte

DataWorks est une plateforme complète de développement et de gouvernance des big data, construite sur des moteurs de traitement de données massives. Elle intègre des fonctionnalités telles que le développement de données, l'ordonnancement des tâches et la gestion des données. Utilisez les tâches de synchronisation de DataWorks pour transférer rapidement les données de diverses sources vers Alibaba Cloud Elasticsearch.

• Sources de données prises en charge :

  • Bases de données Alibaba Cloud (MySQL, PostgreSQL, SQL Server, MongoDB et HBase)

  • Alibaba Cloud PolarDB-X (évolution de DRDS)

  • Alibaba Cloud MaxCompute

  • Alibaba Cloud OSS

  • Alibaba Cloud Tablestore

  • Versions auto-gérées de HDFS, Oracle, FTP, DB2 et autres types de bases de données pris en charge

• Scénarios d'utilisation :

  • Synchronisation hors ligne de big data vers Alibaba Cloud Elasticsearch. Vous pouvez synchroniser une base de données entière ou toutes les données d'une table spécifique. Pour plus d'informations, consultez la rubrique Synchronisation hors ligne d'une base de données MySQL entière vers Elasticsearch.

  • Synchronisation en temps réel de big data vers Alibaba Cloud Elasticsearch. Cette méthode prend en charge à la fois la synchronisation complète et incrémentielle. Pour plus d'informations, consultez la rubrique Synchronisation en temps réel d'une base de données MySQL entière vers Elasticsearch.

Prérequis

• Créez une instance RDS for MySQL. Pour plus d'informations, consultez la rubrique Créer une instance RDS for MySQL. Cette rubrique utilise MySQL 5.7 à titre d'exemple.

• Créez une instance Alibaba Cloud Elasticsearch et activez la fonctionnalité de création automatique d'index. Pour plus d'informations, consultez les rubriques Créer une instance Alibaba Cloud Elasticsearch et Configurer les paramètres YML.

• Créez un espace de travail DataWorks. Pour plus d'informations, consultez la rubrique Créer un espace de travail.

Facturation

• Pour obtenir des informations sur les frais liés à une instance Alibaba Cloud Elasticsearch, consultez la rubrique Éléments facturables d'Elasticsearch.

• Pour obtenir des informations sur les frais liés à un groupe de ressources Data Integration, consultez la rubrique Frais des groupes de ressources.

Étape 1 : Préparer les données source

Créez une base de données et une table dans l'instance RDS for MySQL.

• Vous pouvez utiliser une base de données Alibaba Cloud RDS ou une base de données auto-gérée sur un serveur local. Cette rubrique utilise une base de données RDS for MySQL à titre d'exemple. Pour plus d'informations, consultez la rubrique Démarrage rapide.

• L'extrait de code suivant montre comment créer une table et y insérer des données.

  -- 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');

Étape 2 : Acheter et configurer un groupe de ressources

Achetez un groupe de ressources Data Integration et associez-lui un VPC ainsi qu'un espace de travail.

1. Connectez-vous à la console DataWorks.

2. Dans la barre de navigation supérieure, sélectionnez une région. Dans le volet de navigation de gauche, cliquez sur Resource Groups.

3. Cliquez sur Create Resource Group et configurez les paramètres selon les instructions affichées.

4. Dans la colonne Actions du groupe de ressources créé, cliquez sur Network Settings pour associer un VPC au groupe de ressources. Pour plus d'informations, consultez la rubrique Configurer les paramètres réseau.

   Afin d'autoriser la synchronisation des données, le groupe de ressources doit pouvoir se connecter à la fois à l'instance RDS for MySQL et à l'instance Elasticsearch. Par conséquent, associez le groupe de ressources au même Virtual Private Cloud (VPC), à la même Zone et au même VSwitch que vos instances. Pour trouver les informations VPC de vos instances, consultez les rubriques Changer le VPC et le vSwitch d'une instance RDS for MySQL et Afficher les informations de base d'une instance Elasticsearch.

   Important

   Après avoir associé le VPC, ajoutez le bloc CIDR du vSwitch CIDR Block à la liste d'autorisation IP de l'instance RDS for MySQL et de l'instance Elasticsearch. Pour plus d'informations, consultez les rubriques Configurer une liste d'autorisation IP pour une instance RDS for MySQL et Configurer une liste d'autorisation IP publique ou privée pour une instance Elasticsearch.

5. Dans le coin supérieur gauche de la page, cliquez sur l'icône de retour pour revenir à la page Resource List.

6. Dans la colonne Actions du groupe de ressources créé, cliquez sur Bind workspace pour associer l'espace de travail cible au groupe de ressources.

Étape 3 : Ajouter des sources de données

Ajoutez les sources de données RDS for MySQL et Elasticsearch au service Data Integration de DataWorks.

1. Accédez à la page Data Integration dans DataWorks.

   1. Connectez-vous à la console DataWorks.

   2. Dans le volet de navigation de gauche, cliquez sur Workspace.

   3. Dans la colonne Actions de l'espace de travail cible, choisissez Quick access > Data Integration.

2. Dans le volet de navigation de gauche, cliquez sur Data Source.

3. Ajoutez une source de données RDS for MySQL.

   1. Sur la page Data Sources, cliquez sur Add Data Source.

   2. Sur la page Add Data Source, recherchez et sélectionnez MySQL.

   3. Dans la boîte de dialogue Add MySQL data source, configurez les paramètres de la source de données dans la section Basic Information.

      Pour plus d'informations, consultez la rubrique Configurer une source de données MySQL.

   4. Dans la section Connection Configuration, cliquez sur Test Connectivity. Le statut Connected indique que la connexion a réussi.

   5. Cliquez sur Complete.

4. Ajoutez une source de données Elasticsearch de la même manière. Pour plus d'informations, consultez la rubrique Configurer une source de données Elasticsearch.

Étape 4 : Configurer et exécuter une tâche de synchronisation hors ligne

Une tâche de synchronisation hors ligne s'exécute sur un groupe de ressources, qui sert de ressource de calcul. Le groupe de ressources récupère les données depuis la source de données configurée et les écrit dans Elasticsearch.

1. Accédez à la page Data development dans DataWorks.

   1. Connectez-vous à la console DataWorks.

   2. Dans le volet de navigation de gauche, cliquez sur Workspace.

   3. Dans la colonne Actions de l'espace de travail cible, choisissez Quick access > Data development.

2. Créez un nœud de synchronisation hors ligne.

   1. Dans l'onglet Data Studio (icône ) du volet de navigation de gauche, choisissez Create > New business process. Ensuite, créez un workflow selon les instructions.

   2. Cliquez avec le bouton droit sur le workflow créé et choisissez Create Node > Data Integration > Batch Synchronization.

   3. Dans la boîte de dialogue Create Node, saisissez un nom de nœud et cliquez sur OK.

3. Configurez le réseau et les ressources.

   1. Dans la zone Source, pour Source, sélectionnez MySQL, et pour Data Source Name, sélectionnez le nom de la source de données à synchroniser.

   2. Dans la section Resource Group, sélectionnez un groupe de ressources.

   3. Dans la section Destination, pour Destination, sélectionnez Elasticsearch, et pour Data Source Name, sélectionnez le nom de la source de données à synchroniser.

4. Cliquez sur Next step.

5. Configurez la tâche.

   1. Dans la section Source, sélectionnez la table que vous souhaitez synchroniser.

   2. Dans la section Destination, configurez les paramètres de destination.

   3. Dans la section Field Mapping, configurez les mappages entre Source Column et Target Column.

   4. Dans la section Channel Control, configurez les paramètres du canal.

   Pour plus d'informations sur la configuration, consultez la rubrique Configurer une tâche de synchronisation à l'aide de l'assistant.

6. Exécutez la tâche.

   1. (Facultatif) Configurez les propriétés de planification de la tâche. Dans le volet de droite, cliquez sur Properties et configurez les paramètres de planification selon vos besoins. Pour plus d'informations sur les paramètres, consultez la rubrique Configurer les propriétés de planification.

   2. Dans le coin supérieur gauche de l'éditeur de nœud, cliquez sur l'icône Enregistrer pour sauvegarder la tâche.

   3. Dans le coin supérieur gauche de l'éditeur de nœud, cliquez sur l'icône Soumettre pour soumettre la tâche.

      Si vous avez configuré des propriétés de planification, la tâche s'exécute automatiquement selon le planning défini. Vous pouvez également cliquer sur l'icône Exécuter dans le coin supérieur gauche de l'éditeur de nœud pour lancer immédiatement la tâche.

      Le message Shell run successfully! dans le journal opérationnel indique que la tâche a été exécutée avec succès.

Étape 5 : Vérifier le résultat

1. Connectez-vous à la console Kibana de l'instance Alibaba Cloud Elasticsearch de destination. Pour plus d'informations, consultez la rubrique Se connecter à la console Kibana.

2. Dans le coin supérieur gauche de la page Kibana, cliquez sur l'icône et sélectionnez Dev Tools.

3. Dans la Console, exécutez la commande suivante pour afficher les données synchronisées.

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

   Remarque

   Remplacez es_test par le nom d'index que vous avez spécifié dans la tâche de synchronisation des données.

   Si la synchronisation des données a réussi, le résultat suivant est renvoyé.

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