:Utiliser Logstash pour synchroniser des données d'ApsaraDB RDS for MySQL vers Elasticsearch

Étape 1 : Préparer l'environnement

1. Activez la fonctionnalité Auto Indexing pour votre cluster Elasticsearch afin que Logstash puisse créer automatiquement des index. Pour plus de détails, consultez Accéder et configurer un cluster Elasticsearch.

2. Téléchargez un pilote JDBC compatible avec votre version de MySQL dans le cluster Logstash. Cet exemple utilise mysql-connector-java-5.1.48.jar. Pour plus de détails, consultez Configurer des bibliothèques tierces.

3. Préparez les données de test. L'instruction suivante permet de créer une table :

   CREATE table food (
     id int PRIMARY key AUTO_INCREMENT,
     name VARCHAR (32),
     insert_time DATETIME,
     update_time DATETIME
   );

   L'instruction suivante permet d'insérer des données :

   INSERT INTO food values(null,'Chocolates',now(),now());
   INSERT INTO food values(null,'Yogurt',now(),now());
   INSERT INTO food values(null,'Ham sausage',now(),now());

4. Ajoutez les adresses IP des nœuds Alibaba Cloud Logstash à la liste d'autorisation de l'instance ApsaraDB RDS for MySQL. Vous pouvez obtenir ces adresses IP sur la page Basic Information du cluster Logstash.

Étape 2 : Configurer un pipeline Logstash

1. Accédez à la page Logstash Clusters.

2. Accédez au cluster cible.

   1. Dans la barre de navigation supérieure, sélectionnez la région où réside le cluster.

   2. Sur la page Logstash Clusters, localisez le cluster et cliquez sur son ID.

3. Dans le volet de navigation de gauche, cliquez sur Pipelines.

4. Cliquez sur Create Pipeline.

5. Sur la page Create Pipeline, saisissez un Pipeline ID et configurez la section Config.

   La configuration Config suivante est utilisée dans cette rubrique.

   input {
     jdbc {
       jdbc_driver_class => "com.mysql.jdbc.Driver"
       jdbc_driver_library => "/ssd/1/share/<Logstash cluster ID>/logstash/current/config/custom/mysql-connector-java-5.1.48.jar"
       jdbc_connection_string => "jdbc:mysql://rm-bp1xxxxx.mysql.rds.aliyuncs.com:3306/<Database name>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false"
       jdbc_user => "xxxxx"
       jdbc_password => "xxxx"
       jdbc_paging_enabled => "true"
       jdbc_page_size => "50000"
       statement => "select * from food where update_time >= :sql_last_value"
       schedule => "* * * * *"
       record_last_run => true
       last_run_metadata_path => "/ssd/1/<Logstash cluster ID>/logstash/data/last_run_metadata_update_time.txt"
       clean_run => false
       tracking_column_type => "timestamp"
       use_column_value => true
       tracking_column => "update_time"
     }
   }
   filter {
   }
   output {
     elasticsearch {
       hosts => "http://es-cn-0h****dd0hcbnl.elasticsearch.aliyuncs.com:9200"
       index => "rds_es_dxhtest_datetime"
       user => "elastic"
       password => "xxxxxxx"
       document_id => "%{id}"
     }
   }

   Remarque

   Remplacez l'espace réservé <Logstash cluster ID> dans le code par l'ID du cluster Logstash que vous avez créé. Pour savoir comment obtenir cet ID, consultez Afficher les informations de base d'un cluster.

   Configuration	Description
   input	Spécifie la source de données d'entrée. Pour connaître les types de sources de données pris en charge, consultez Plug-ins d'entrée. Cette rubrique utilise une source de données JDBC. Pour la description des paramètres, consultez paramètres d'entrée.
   filter	Spécifie le plug-in utilisé pour filtrer les données d'entrée. Pour connaître les types de plug-ins pris en charge, consultez Plug-ins de filtre.
   output	Spécifie le type de destination des données. Pour connaître les types de destinations pris en charge, consultez Plug-ins de sortie. Dans cette rubrique, les données de MySQL sont synchronisées vers Elasticsearch. Par conséquent, les informations relatives au cluster Elasticsearch de destination doivent être spécifiées dans output. Pour la description des paramètres, consultez Étape 3 : Créer et exécuter un pipeline. Important Si le paramètre file_extend est utilisé dans output, le plug-in logstash-output-file_extend doit être installé au préalable. Pour plus d'informations, consultez Installer ou supprimer des plug-ins.

   Paramètre	Description
   jdbc_driver_class	Configuration de la classe JDBC.
   jdbc_driver_library	Spécifie le fichier de pilote JDBC utilisé pour se connecter à MySQL. Le format est /ssd/1/share/<Logstash cluster ID>/logstash/current/config/custom/<nom du fichier de pilote>. Vous devez télécharger le fichier de pilote dans la console au préalable. Pour connaître les fichiers de pilote pris en charge par Alibaba Cloud Logstash et la procédure de téléchargement, consultez Configurer des bibliothèques tierces.
   jdbc_connection_string	Spécifie le nom de domaine, le port et la base de données de la connexion. Le format est jdbc:mysql://<point de terminaison MySQL>:<Port>/<Nom de la base de données>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false. <point de terminaison MySQL> : spécifiez le point de terminaison interne de MySQL. Si vous utilisez le point de terminaison public, vous devez configurer une passerelle NAT pour Logstash et définir jdbc:mysql://<point de terminaison MySQL>:<Port> avec le nom de domaine public afin que les données soient transmises via Internet. Pour plus d'informations, consultez Configurer une passerelle NAT pour le transfert de données publiques. <Port> : le port doit correspondre au port sortant de MySQL. En général, il s'agit du port 3306.
   jdbc_user	Nom d'utilisateur de la base de données.
   jdbc_password	Mot de passe de la base de données.
   jdbc_paging_enabled	Indique si la pagination est activée. Valeur par défaut : false.
   jdbc_page_size	Taille de la pagination.
   statement	Spécifie l'instruction SQL. Pour une requête portant sur plusieurs tables, vous pouvez utiliser une instruction JOIN. Remarque sql_last_value sert à déterminer la ligne à interroger. Avant toute exécution de requête, cette valeur est définie au jeudi 1er janvier 1970. Pour plus d'informations, consultez Plug-in d'entrée Jdbc.
   schedule	Spécifie l'opération planifiée. La valeur * indique que les données sont synchronisées toutes les minutes. Ce paramètre utilise une expression cron de style Rufus.
   record_last_run	Indique si le résultat de l'exécution précédente doit être enregistré. Si ce paramètre est défini sur true, la valeur du champ tracking_column de l'exécution précédente est enregistrée et sauvegardée dans le fichier spécifié par last_run_metadata_path.
   last_run_metadata_path	Spécifie le chemin du fichier stockant l'heure de la dernière exécution. Le backend expose actuellement le chemin /ssd/1/<Logstash cluster ID>/logstash/data/ pour stocker le fichier. Après avoir spécifié le chemin, Logstash génère automatiquement un fichier à cet emplacement, mais le contenu du fichier n'est pas visible. Remarque Lors de la configuration d'un pipeline Logstash, nous vous recommandons de définir ce paramètre en utilisant le chemin /ssd/1/<Logstash cluster ID>/logstash/data/. Si vous n'utilisez pas ce chemin, les enregistrements conditionnels synchronisés ne pourront pas être stockés dans le fichier de configuration sous le chemin last_run_metadata_path en raison de permissions insuffisantes.
   clean_run	Indique si les enregistrements de last_run_metadata_path doivent être effacés. Valeur par défaut : false. Si ce paramètre est défini sur true, tous les enregistrements de la base de données sont interrogés depuis le début à chaque exécution.
   use_column_value	Indique si la valeur d'une colonne doit être enregistrée. Si ce paramètre est défini sur true, le système enregistre la dernière valeur de la colonne spécifiée par tracking_column et utilise cette valeur lors de la prochaine exécution du pipeline pour déterminer les enregistrements à mettre à jour.
   tracking_column_type	Type de la colonne de suivi. Valeur par défaut : numeric.
   tracking_column	Spécifie la colonne de suivi. La colonne doit être incrémentielle. Il s'agit généralement de la clé primaire MySQL.

   Important

   • La configuration précédente repose sur les données de test. Dans des scénarios professionnels réels, configurez les paramètres selon vos besoins métier. Pour les autres options de configuration prises en charge par le plug-in d'entrée, consultez la rubrique officielle Plug-in d'entrée Logstash Jdbc.

   • Si la configuration contient un paramètre tel que last_run_metadata_path, Alibaba Cloud Logstash doit fournir le chemin du fichier. Le backend expose actuellement le chemin /ssd/1/<Logstash cluster ID>/logstash/data/ à des fins de test, et les données de ce répertoire ne sont pas supprimées. Assurez-vous donc que le disque dispose d'un espace disponible suffisant. Après avoir spécifié le chemin, Logstash génère automatiquement un fichier à cet emplacement, mais le contenu du fichier n'est pas visible.

   • Pour renforcer la sécurité, si un pilote JDBC est utilisé lors de la configuration du pipeline, vous devez ajouter allowLoadLocalInfile=false&autoDeserialize=false au paramètre jdbc_connection_string. Sinon, le système de planification signale un échec de vérification lors de l'ajout du fichier de configuration Logstash. Exemple : jdbc_connection_string => "jdbc:mysql://xxx.drds.aliyuncs.com:3306/<Database name>?allowLoadLocalInfile=false&autoDeserialize=false".

   Pour plus d'exemples de configurations Config, consultez Fichiers de configuration Logstash.

6. Cliquez sur Next pour configurer les paramètres du pipeline.

   Avertissement

   La sauvegarde et le déploiement du pipeline déclenchent un redémarrage du cluster Logstash. Confirmez qu'un redémarrage n'affectera pas vos charges de travail avant de poursuivre.

   Paramètre	Description	Par défaut
   Pipeline Workers	Nombre de threads exécutant les plug-ins de filtre et de sortie en parallèle. Augmentez cette valeur lorsque les ressources CPU sont sous-utilisées ou lorsque les événements s'accumulent.	Nombre de vCPU
   Pipeline Batch Size	Nombre maximal d'événements collectés par un seul worker à partir des entrées avant l'exécution des filtres et des sorties. Des valeurs plus élevées augmentent le débit, mais nécessitent davantage de mémoire JVM heap.	125
   Pipeline Batch Delay	Durée d'attente d'un worker pour des événements supplémentaires avant de démarrer un petit lot, en millisecondes.	50
   Queue Type	Modèle de file d'attente interne pour la mise en tampon des événements. MEMORY : file d'attente en mémoire. PERSISTED : file d'attente basée sur le disque avec accusé de réception pour la durabilité.	MEMORY
   Queue Max Bytes	Taille maximale de la file d'attente sur le disque. Doit être inférieure à la capacité disponible de votre disque.	1024 Mo
   Queue Checkpoint Writes	Nombre maximal d'événements écrits avant qu'un point de contrôle ne soit forcé (files d'attente persistantes uniquement). Définissez sur 0 pour aucune limite.	1024

   

7. Cliquez sur Save and Deploy pour redémarrer le cluster Logstash et appliquer immédiatement la configuration. Vous pouvez également cliquer sur Save pour enregistrer la configuration et déclencher une modification du cluster, mais les paramètres ne prendront pas effet immédiatement. Pour déployer ultérieurement, accédez à la page Pipelines dans la colonne Deploy Now.

Étape 3 : Vérifier le résultat

1. Connectez-vous à la console Kibana de votre cluster Elasticsearch. Pour plus d'informations, consultez la rubrique Se connecter à la console Kibana.

2. Dans le coin supérieur gauche, cliquez sur l'icône et choisissez Actions > Management.

3. Sous l'onglet Dev Tools, exécutez la commande suivante pour confirmer que les trois lignes ont été synchronisées :

   GET rds_es_dxhtest_datetime/_count
   {
     "query": {"match_all": {}}
   }

   Réponse attendue :

   {
     "count" : 3,
     "_shards" : {
       "total" : 1,
       "successful" : 1,
       "skipped" : 0,
       "failed" : 0
     }
   }

4. Mettez à jour et insérez des lignes dans MySQL pour tester la synchronisation incrémentielle :

   UPDATE food SET name='Chocolates',update_time=now() where id = 1;
   INSERT INTO food values(null,'Egg',now(),now());

5. Dans la console Kibana, consultez les données mises à jour.

   • Recherchez la ligne mise à jour :

     GET rds_es_dxhtest_datetime/_search
     {
       "query": {
         "match": {
           "name": "Chocolates"
         }
       }
     }

     
     {
       "took" : 2,
       "timed_out" : false,
       "_shards" : {
         "total" : 1,
         "successful" : 1,
         "skipped" : 0,
         "failed" : 0
       },
       "hits" : {
         "total" : {
           "value" : 1,
           "relation" : "eq"
         },
         "max_score" : 1.5580825,
         "hits" : [
           {
             "_index" : "rds_es_dxhtest_datetime",
             "_type" : "_doc",
             "_id" : "1",
             "_score" : 1.5580825,
             "_source" : {
               "update_time" : "2020-03-23T03:43:19.000Z",
               "@version" : "1",
               "name" : "Chocolates",
               "insert_time" : "2020-03-23T03:00:36.000Z",
               "@timestamp" : "2020-03-23T03:44:00.1857",
               "id" : 1
             }
           }
         ]
       }
     }

   • Interrogez tous les documents :

     GET rds_es_dxhtest_datetime/_search
     {
       "query": {
         "match_all": {}
       }
     }

     La réponse inclut le document nouvellement ajouté avec id=4 et name=Egg, ce qui confirme que la synchronisation des données a réussi.

     
     {
       "_index" : "rds_es_dxhtest_datetime",
       "_type" : "_doc",
       "_id" : "1",
       "_score" : 1.0,
       "_source" : {
         "update_time" : "2020-03-23T03:43:19.000Z",
         "@version" : "1",
         "name" : "Chocolates",
         "insert_time" : "2020-03-23T03:00:36.000Z",
         "@timestamp" : "2020-03-23T03:44:00.185Z",
         "id" : 1
       }
     },
     {
       "_index" : "rds_es_dxhtest_datetime",
       "_type" : "_doc",
       "_id" : "4",
       "_score" : 1.0,
       "_source" : {
         "update_time" : "2020-03-23T04:05:01.000Z",
         "@version" : "1",
         "name" : "Egg",
         "insert_time" : "2020-03-23T04:05:01.000Z",
         "@timestamp" : "2020-03-23T04:06:00.192Z",
         "id" : 4
       }
     }

input {
    jdbc {
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_driver_library => "/ssd/1/share/<Logstash cluster ID>/logstash/current/config/custom/mysql-connector-java-5.1.48.jar"
      jdbc_connection_string => "jdbc:mysql://rm-bp1xxxxx.mysql.rds.aliyuncs.com:3306/<Database name>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false"
      jdbc_user => "xxxxx"
      jdbc_password => "xxxx"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      statement => "select * from tableA where update_time >= :sql_last_value"
      schedule => "* * * * *"
      record_last_run => true
      last_run_metadata_path => "/ssd/1/<Logstash cluster ID>/logstash/data/last_run_metadata_update_time.txt"
      clean_run => false
      tracking_column_type => "timestamp"
      use_column_value => true
      tracking_column => "update_time"
      type => "A"
    }
    jdbc {
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_driver_library => "/ssd/1/share/<Logstash cluster ID>/logstash/current/config/custom/mysql-connector-java-5.1.48.jar"
      jdbc_connection_string => "jdbc:mysql://rm-bp1xxxxx.mysql.rds.aliyuncs.com:3306/<Database name>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false"
      jdbc_user => "xxxxx"
      jdbc_password => "xxxx"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      statement => "select * from tableB where update_time >= :sql_last_value"
      schedule => "* * * * *"
      record_last_run => true
      last_run_metadata_path => "/ssd/1/<Logstash cluster ID>/logstash/data/last_run_metadata_update_time.txt"
      clean_run => false
      tracking_column_type => "timestamp"
      use_column_value => true
      tracking_column => "update_time"
      type => "B"
    }
}
output {
    if[type] == "A" {
        elasticsearch {
            hosts => "http://es-cn-0h****dd0hcbnl.elasticsearch.aliyuncs.com:9200"
            index => "rds_es_dxhtest_datetime_A"
            user => "elastic"
            password => "xxxxxxx"
            document_id => "%{id}"
        }
    }
    if[type] == "B" {
        elasticsearch {
            hosts => "http://es-cn-0h****dd0hcbnl.elasticsearch.aliyuncs.com:9200"
            index => "rds_es_dxhtest_datetime_B"
            user => "elastic"
            password => "xxxxxxx"
            document_id => "%{id}"
        }
    }
}