Synchroniser des données de PolarDB-X (DRDS) vers Elasticsearch à l'aide de Logstash - Elasticsearch

Contexte

Alibaba Cloud Logstash est un outil puissant de collecte et de traitement des données. Il permet d'ingérer, de transformer, d'optimiser et de restituer les données. Le plug-in logstash-input-jdbc, installé par défaut et non supprimable, interroge les données de PolarDB-X par lots et les synchronise vers Elasticsearch. Ce plug-in effectue également des interrogations périodiques sur PolarDB-X afin de synchroniser les enregistrements insérés ou modifiés depuis la dernière interrogation. Cette solution convient parfaitement à une synchronisation initiale complète ainsi qu'aux mises à jour incrémentielles ultérieures, qui présentent généralement une latence de quelques secondes.

Prérequis

Créez une instance PolarDB-X 1,0, un cluster Alibaba Cloud Elasticsearch et un cluster Alibaba Cloud Logstash. Créez également une base de données dans l'instance PolarDB-X 1,0. Nous vous recommandons de créer l'instance PolarDB-X 1,0, le cluster Elasticsearch et le cluster Logstash au sein du même réseau privé virtuel (VPC).

Une instance PolarDB-X 1,0 et une base de données ont été créées. Pour plus d'informations, consultez Créer une instance PolarDB-X 1,0.
Pour savoir comment créer un cluster Alibaba Cloud Elasticsearch, consultez la rubrique Créer un cluster Alibaba Cloud Elasticsearch. Dans cet exemple, un cluster Elasticsearch V6,7 de l'édition Standard est créé.
Pour savoir comment créer un cluster Alibaba Cloud Logstash, consultez la rubrique Créer un cluster Alibaba Cloud Logstash.

Remarque
Si vous souhaitez utiliser Logstash pour collecter des données depuis Internet ou transférer les données collectées vers Internet, configurez une passerelle NAT (Network Address Translation) et utilisez-la pour connecter votre cluster Logstash à Internet. Pour plus d'informations, consultez la rubrique Configurer une passerelle NAT pour la transmission de données via Internet.

Limites

Les valeurs du champ _id dans le cluster Elasticsearch doivent être identiques aux valeurs du champ id dans la base de données PolarDB-X 1,0.

Cette condition garantit que la tâche de synchronisation des données établit une correspondance entre les enregistrements de la base de données PolarDB-X 1,0 et les documents du cluster Elasticsearch. Si vous mettez à jour un enregistrement dans la base de données PolarDB-X 1,0, la tâche de synchronisation utilise l'enregistrement mis à jour pour écraser le document portant le même ID dans le cluster Elasticsearch.

Remarque
En substance, une opération de mise à jour dans Elasticsearch supprime le document original et indexe le nouveau document. Par conséquent, l'opération d'écrasement est aussi efficace qu'une opération de mise à jour effectuée par la tâche de synchronisation des données.
Si vous insérez ou mettez à jour un enregistrement dans la base de données PolarDB-X 1,0, cet enregistrement doit contenir un champ indiquant l'heure d'insertion ou de modification.

À chaque cycle d'interrogation, le plug-in logstash-input-jdbc enregistre l'heure d'insertion ou de modification du dernier enregistrement dans la base de données PolarDB-X 1,0. Logstash ne synchronise depuis la base de données PolarDB-X 1,0 que les enregistrements répondant aux critères suivants : l'heure d'insertion ou de modification des enregistrements dans PolarDB-X 1,0 est postérieure à l'heure d'insertion ou de modification du dernier enregistrement du cycle précédent.

Important
Si vous supprimez des enregistrements dans la base de données PolarDB-X 1,0, le plug-in logstash-input-jdbc ne peut pas supprimer les documents correspondants (ayant les mêmes ID) du cluster Elasticsearch. Pour supprimer ces documents du cluster Elasticsearch, exécutez la commande appropriée directement sur le cluster Elasticsearch.

Procédure

Étape 1 : Préparer l'environnement

Créez une table dans votre instance PolarDB-X 1,0 et remplissez-la avec des données de test.

Dans cet exemple, la table est créée à l'aide de l'instruction suivante :

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

Les instructions suivantes permettent d'insérer des données dans la table :

INSERT INTO food values(null,'chocolate',now(),now());
INSERT INTO food values(null,'yogurt',now(),now());
INSERT INTO food values(null,'ham sausage',now(),now());

Activez la fonctionnalité Indexation automatique pour le cluster Elasticsearch. Pour plus d'informations, consultez la rubrique Accéder et configurer un cluster Elasticsearch.
Dans le cluster Logstash, téléchargez un pilote JDBC SQL dont la version est compatible avec celle de la base de données PolarDB-X 1,0. Pour plus d'informations, consultez la rubrique Configurer des bibliothèques tierces. Dans cet exemple, le pilote mysql-connector-java-5.1.35 est utilisé.

Remarque
Dans cet exemple, un pilote JDBC MySQL est utilisé pour se connecter à la base de données PolarDB-X 1,0. Vous pouvez également utiliser un pilote JDBC PolarDB. Toutefois, pour une base de données PolarDB-X 2,0, un pilote JDBC PolarDB risque de ne pas fonctionner. Nous vous recommandons d'utiliser un pilote JDBC MySQL.
Récupérez les adresses IP des nœuds du cluster Logstash sur la page Informations de base du cluster Logstash dans la console Elasticsearch. Ajoutez ensuite ces adresses IP à la liste d'autorisation d'adresses IP de l'instance PolarDB-X 1,0. Pour plus d'informations, consultez la rubrique Définir une liste d'autorisation d'adresses IP.

Étape 2 : Configurer un pipeline Logstash

Accédez à la page Clusters Logstash.
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.
Dans le volet de navigation de gauche, cliquez sur Pipelines.
Cliquez sur Create Pipeline.

Sur la page Create, saisissez un Pipeline ID et configurez le pipeline dans la zone de texte Config.

Dans cet exemple, les configurations suivantes sont saisies dans le champ Config Settings :

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.35.jar"
    jdbc_connection_string => "jdbc:mysql://drdshbga51x6****.drds.aliyuncs.com:3306/<Your database name>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false"
    jdbc_user => "db_user"
    jdbc_password => "db_password"
    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-n6w1o1x0w001c****.elasticsearch.aliyuncs.com:9200"
    user => "elastic"
    password => "es_password"
    index => "drds_test"
    document_id => "%{id}"
  }
}

Remarque

Dans le code, remplacez <Logstash cluster ID> par l'ID de votre cluster Logstash. Pour plus d'informations, consultez la rubrique Afficher la vue d'ensemble de la liste des clusters.

Tableau 1. Paramètres de la section input

Paramètre

Description

jdbc_driver_class

La classe du pilote JDBC.

jdbc_driver_library

Le chemin d'accès au fichier du pilote JDBC. Pour plus d'informations, consultez la rubrique Configurer des bibliothèques tierces.

jdbc_connection_string

La chaîne de connexion JDBC utilisée pour se connecter à la base de données PolarDB-X 1,0. Cette chaîne contient le point de terminaison, le numéro de port et le nom de la base de données PolarDB-X 1,0.

jdbc_user

Le nom d'utilisateur utilisé pour accéder à la base de données PolarDB-X 1,0.

jdbc_password

Le mot de passe utilisé pour accéder à la base de données PolarDB-X 1,0.

jdbc_paging_enabled

Indique si la pagination est activée. Valeur par défaut : false.

jdbc_page_size

Le nombre d'entrées par page.

statement

L'instruction SQL.

schedule

La planification de la tâche de synchronisation. "* * * * *" signifie que les données sont synchronisées toutes les minutes.

record_last_run

Indique si le dernier résultat d'exécution doit être enregistré. Si ce paramètre est défini sur true, la valeur de tracking_column du dernier résultat d'exécution est stockée dans le fichier situé dans le chemin spécifié par le paramètre last_run_metadata_path.

last_run_metadata_path

Le chemin d'accès au fichier contenant la dernière heure d'exécution. Un chemin d'accès est fourni en arrière-plan. Le format du chemin est /ssd/1/<ID du cluster Logstash>/logstash/data/. Après avoir spécifié un chemin, Logstash génère automatiquement un fichier à cet emplacement, mais vous ne pouvez pas afficher les données contenues dans le fichier.

clean_run

Indique si le chemin spécifié par le paramètre last_run_metadata_path doit être effacé. Valeur par défaut : false. Si ce paramètre est défini sur true, chaque interrogation commence à partir de la première entrée de la base de données.

use_column_value

Indique si les valeurs d'une colonne spécifique doivent être enregistrées.

tracking_column_type

Le type de la colonne dont vous souhaitez suivre les valeurs. Valeur par défaut : numeric.

tracking_column

La colonne dont vous souhaitez suivre les valeurs. Les valeurs doivent être triées par ordre croissant. En général, cette colonne correspond à la clé primaire de la table.

Tableau 2. Paramètres de la section output

Paramètre

Description

hosts

Le point de terminaison privé du cluster Alibaba Cloud Elasticsearch, au format http://<Point de terminaison privé du cluster>:9200. Vous trouverez le point de terminaison privé sur la page Informations de base. Pour plus d'informations, consultez la rubrique Afficher les informations de base d'un cluster.

user

Le nom d'utilisateur utilisé pour accéder au cluster Elasticsearch. Le nom d'utilisateur par défaut est elastic.

password

Le mot de passe du compte elastic. Ce mot de passe est spécifié lors de la création du cluster Elasticsearch. Si vous l'avez oublié, vous pouvez le réinitialiser. Pour connaître la procédure et les précautions liées à la réinitialisation du mot de passe, consultez la rubrique Réinitialiser le mot de passe d'accès d'un cluster Elasticsearch.

index

Le nom de l'index dans le cluster Elasticsearch.

document_id

Les ID des documents dans le cluster Elasticsearch. Définissez ce paramètre sur %{id}, ce qui indique que les ID des documents sont identiques aux ID des enregistrements de données dans la base de données PolarDB-X 1,0.

Important

Les configurations précédentes sont basées sur des données de test. Vous pouvez configurer le pipeline selon vos besoins métier. Pour plus d'informations sur les autres paramètres pris en charge par le plug-in d'entrée, consultez la documentation Plug-in d'entrée Jdbc de Logstash.
Si votre configuration inclut des paramètres tels que last_run_metadata_path, le chemin d'accès au fichier doit être fourni par le service Alibaba Cloud Logstash. Le système fournit le chemin /ssd/1/<Logstash cluster ID>/logstash/data/ à des fins de test. Les données de ce répertoire ne sont pas supprimées. Assurez-vous de disposer d'un espace disque suffisant. Logstash crée automatiquement un fichier au chemin spécifié, mais vous ne pouvez pas consulter son contenu.
Pour renforcer la sécurité, si vous utilisez un pilote JDBC dans la configuration de votre pipeline, ajoutez allowLoadLocalInfile=false&autoDeserialize=false au paramètre jdbc_connection_string. Sinon, le système de planification signalera un échec de validation lors de l'ajout du fichier de configuration Logstash. Exemple : jdbc_connection_string => "jdbc:mysql://drdshbga51x6****.drds.aliyuncs.com:3306/<Your database name>?allowLoadLocalInfile=false&autoDeserialize=false".

Pour plus d'informations sur la configuration des paramètres dans le champ Config Settings, consultez la rubrique Fichiers de configuration Logstash.

Cliquez sur Next step et configurez les paramètres du pipeline.

Paramètre	Description
Pipeline Workers	Nombre de threads de travail pour les étapes de filtrage et de sortie. Augmentez cette valeur si les événements sont mis en attente ou si le processeur est sous-utilisé. Valeur par défaut : le nombre de cœurs de CPU.
Pipeline Batch Size	Nombre maximal d'événements qu'un worker collecte avant d'exécuter les filtres et les sorties. Des lots plus importants augmentent l'utilisation de la mémoire et peuvent nécessiter une taille de tas JVM plus grande (LS_HEAP_SIZE). Valeur par défaut : 125.
Pipeline Batch Delay	Temps d'attente en millisecondes avant l'envoi d'un lot de taille insuffisante à un thread de travail. Valeur par défaut : 50 ms.
Queue Type	Modèle de file d'attente interne pour la mise en buffer des événements entre les étapes. Valeurs valides : MEMORY : Valeur par défaut. Utilise une file d'attente en mémoire. PERSISTED : File d'attente persistante basée sur le disque.
Queue Max Bytes	Quantité maximale de données que la file d'attente peut stocker, en `Mo`. La valeur doit être un entier compris entre `1` et `2⁵³-1`. Valeur par défaut : `1024`. Remarque Assurez-vous que cette valeur est inférieure à votre capacité totale de disque.
Queue Checkpoint Writes	Nombre maximal d'événements écrits avant qu'un point de contrôle ne soit forcé (file d'attente persistante uniquement). 0 signifie aucune limite. Valeur par défaut : 1024.

Avertissement

L'enregistrement et le déploiement de la configuration déclenchent un redémarrage de l'instance. Procédez uniquement si cela n'affecte pas votre activité.

Cliquez sur Save ou sur Save and Deploy.
- Save : Enregistre la configuration du pipeline sans l'appliquer. Après l'enregistrement, vous revenez à la page Pipelines. Dans la section Pipelines, vous pouvez cliquer sur Deploy Now dans la colonne Actions pour redémarrer l'instance et appliquer la configuration.
- Save and Deploy : Enregistre et déploie la configuration, en redémarrant l'instance pour appliquer les modifications.

Étape 3 : Vérifier le résultat

Connectez-vous à la console Kibana du cluster Elasticsearch.

Pour plus d'informations, consultez la rubrique Se connecter à la console Kibana.
Dans le volet de navigation de gauche, cliquez sur Dev Tools.
Dans la Console, exécutez la commande suivante pour afficher le nombre de documents synchronisés.
```
GET drds_test/_count
{
  "query": {"match_all": {}}
}
```
Si la commande s'exécute correctement, le résultat suivant est renvoyé :
```
{
  "count" : 3,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  }
}
```

Mettez à jour les données de la table et insérez-y de nouvelles données.

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

Consultez les données mises à jour et insérées.
- Interrogez l'enregistrement de données dont la valeur de name est Chocolates.
```
GET drds_test/_search
{
  "query": {
    "match": {
      "name": "Chocolates"
   }}
}
```
  Si la commande s'exécute correctement, le résultat illustré dans la figure ci-dessous est renvoyé.
- Interrogez toutes les données.
```
GET drds_test/_search
{
  "query": {
    "match_all": {}
  }
}
```
  Si la commande s'exécute correctement, le résultat illustré dans la figure ci-dessous est renvoyé.

FAQ

FAQ sur le transfert de données à l'aide de Logstash