Réduisez, réutilisez, recyclez (votre code). Apprenez à créer des composants modulaires pour vos pipelines d'ingestion Logstash.

<p><a href="/fr/products/logstash">Logstash</a> est un pipeline open&nbsp;source destiné au traitement des données. Sa mission&nbsp;? Ingérer des événements depuis une ou plusieurs entrées, les transformer, puis envoyer chacun d'eux vers une ou plusieurs sorties. Certaines implémentations Logstash peuvent avoir plusieurs lignes de code et peuvent traiter les événements depuis différentes sources d'entrée. Afin de faciliter la gestion de telles implémentations, je vais vous montrer comment créer des pipelines à partir de composants modulaires, ce qui rend le code plus facilement réutilisable.
</p> <strong><h2>Motivation</h2></strong><p>Logstash doit souvent appliquer un sous-ensemble logique commun à des événements provenant de différentes sources d'entrée. Pour ce faire, on adopte généralement l'une de ces deux approches&nbsp;:&nbsp;
</p><ol>
	<li>Traiter les événements provenant de différentes sources d'entrée dans un seul pipeline, afin de faciliter l'application d'une logique commune à tous les événements, quelle que soit leur source. Dans de telles implémentations, outre la logique commune, on trouve aussi habituellement un nombre considérable de logiques conditionnelles. Par conséquent, cette approche peut se traduire par des implémentations Logstash complexes et difficiles à comprendre.</li>
	<li>Exécuter un pipeline unique pour le traitement des événements provenant de chaque source d'entrée unique. Cette approche nécessite de dupliquer et de copier dans chaque pipeline une fonctionnalité commune, ce qui complique la gestion des parties de code communes.&nbsp;</li>
</ol><p>La méthode présentée dans cet article de blog répond aux inconvénients des deux approches ci-dessus&nbsp;: elle permet de stocker des composants de pipeline modulaires dans différents fichiers, puis d'associer ces composants pour créer des pipelines. Cette méthode peut vous permettre de simplifier le pipeline et vous éviter la duplication de code.
</p> <strong><h2>Création d'un pipeline modulaire</h2></strong><p>Un <a href="/guide/en/logstash/current/configuration.html">fichier de configuration</a> Logstash est composé d'entrées, de filtres et de sorties, qui sont exécutés par un pipeline Logstash.&nbsp; Dans les configurations avancées, on trouve assez souvent une <a href="/guide/en/logstash/current/multiple-pipelines.html">instance Logstash exécutant plusieurs pipelines</a>. Par défaut, lorsque vous lancez Logstash sans argument, il lit un fichier nommé <code>pipelines.yml</code> et instancie les pipelines spécifiés.&nbsp;
</p><p>Vous pouvez stocker les entrées, les filtres et les sorties Logstash dans différents fichiers, que vous pouvez choisir d'inclure dans un pipeline en spécifiant une <a href="/guide/en/logstash/current/glob-support.html">expression&nbsp;glob</a>. Les fichiers qui correspondent à une expression glob sont alors associés dans l'ordre alphabétique. L'ordre d'exécution des filtres étant souvent important, il peut être utile d'inclure des identifiants numériques dans les noms des fichiers, afin qu'ils soient associés dans l'ordre voulu.
</p><p>Ci-dessous, nous allons définir deux pipelines uniques, qui sont en fait l'association de plusieurs composants Logstash modulaires. Nous allons stocker nos composants Logstash components dans les fichiers suivants&nbsp;:
</p><ul>
	<li>Déclarations d'entrée&nbsp;: <code>01_in.cfg</code>, <code>02_in.cfg</code></li>
	<li>Déclarations de filtre&nbsp;:<code> 01_filter.cfg</code>, <code>02_filter.cfg</code>, <code>03_filter.cfg</code></li>
	<li>Déclarations de sortie&nbsp;:<code> 01_out.cfg</code></li>
</ul><p>Au moyen d'<a href="/guide/en/logstash/current/glob-support.html">expressions glob</a>, nous allons ensuite définir les pipelines dans <code>pipelines.yml</code>, pour qu'ils comprennent les composants voulus, comme suit&nbsp;:&nbsp;
</p><pre class="prettyprint">- pipeline.id: my-pipeline_1
  path.config: "&lt;path&gt;/{01_in,01_filter,02_filter,01_out}.cfg"
- pipeline.id: my-pipeline_2
  path.config: "&lt;path&gt;/{02_in,02_filter,03_filter,01_out}.cfg"
</pre><p>Dans la configuration de pipelines ci-dessus, le fichier <code>02_filter.cfg</code> est présent dans les deux pipelines. On voit donc que le code commun aux deux pipelines peut être défini et géré dans un seul fichier, tout en s'exécutant sur plusieurs pipelines.
</p> <strong><h2>Tester les pipelines</h2></strong><p>Dans cette partie, nous allons prendre un exemple concret pour illustrer les fichiers qui seront associés dans les pipelines uniques que nous avons définis ci-dessus dans <code>pipelines.yml</code>. Nous lancerons ensuite Logstash avec ces fichiers et verrons le résultat ainsi généré.&nbsp;
</p> <strong><h3>Fichiers de configuration</h3></strong><p><strong>Fichier d'entrée&nbsp;: 01_in.cfg</strong>
</p><p>L'entrée que définit ce fichier est un <a href="/guide/en/logstash/current/plugins-inputs-generator.html">générateur</a>. L'entrée générateur sert à tester Logstash et dans cet exemple, elle va générer un seul événement.&nbsp;
</p><pre class="prettyprint">input { 
  generator { 
    lines =&gt; ["Generated line"] 
    count =&gt; 1 
  } 
}
</pre><p><strong>Fichier d'entrée&nbsp;: 02_in.cfg</strong>
</p><p>Ce fichier définit une entrée Logstash qui écoute sur&nbsp;stdin&nbsp;.
</p><pre class="prettyprint">input { 
  stdin {} 
}
</pre><p><strong>Fichier de filtre&nbsp;: 01_filter.cfg</strong>
</p><pre class="prettyprint">filter { 
  mutate { 
    add_field =&gt; { "filter_name" =&gt; "Filter 01" } 
  } 
}
</pre><p><strong>Fichier de filtre&nbsp;: 02_filter.cfg</strong>
</p><pre class="prettyprint">filter { 
  mutate { 
    add_field =&gt; { "filter_name" =&gt; "Filter 02" } 
  } 
}
</pre><p><strong>Fichier de filtre&nbsp;: 03_filter.cfg</strong>
</p><pre class="prettyprint">filter { 
  mutate { 
    add_field =&gt; { "filter_name" =&gt; "Filter 03" } 
  } 
}
</pre><p><strong>Fichier de sortie&nbsp;: 01_out.cfg</strong>
</p><pre class="prettyprint">output { 
  stdout { codec =&gt;  "rubydebug" } 
}
</pre> <strong><h3>Exécuter le pipeline</h3></strong><p>Lorsqu'on démarre Logstash sans définir aucune option, on exécute le fichier <code>pipelines.yml</code> défini précédemment. Exécuter Logstash comme suit&nbsp;:&nbsp;
</p><pre class="prettyprint">./bin/logstash
</pre><p>Comme le pipeline <code>my-pipeline_1</code> exécute un générateur pour simuler un événement d'entrée, nous devons voir le résultat suivant aussitôt l'initialisation de Logstash terminée. Nous voyons que ce pipeline exécute les contenus de <code>01_filter.cfg</code> et <code>02_filter.cfg</code> comme prévu.
</p><pre class="prettyprint">{ 
       "sequence" =&gt; 0, 
           "host" =&gt; "alexandersmbp2.lan", 
        "message" =&gt; "Generated line", 
     "@timestamp" =&gt; 2020-02-05T22:10:09.495Z, 
       "@version" =&gt; "1", 
    "filter_name" =&gt; [ 
        [0] "Filter 01", 
        [1] "Filter 02" 
    ] 
}
</pre><p>Étant donné que l'autre pipeline nommé <code>my-pipeline_2</code> attend une entrée sur&nbsp;stdin, nous voyons qu'il n'a encore traité aucun événement. Saisissez quelque chose dans le terminal sur lequel s'exécute Logstash et appuyez sur la touche Retour pour créer un événement pour ce pipeline. Une fois cela fait, le résultat qui s'affiche doit ressembler à ceci&nbsp;:&nbsp;
</p><pre class="prettyprint">{ 
    "filter_name" =&gt; [ 
        [0] "Filter 02", 
        [1] "Filter 03" 
    ], 
           "host" =&gt; "alexandersmbp2.lan", 
        "message" =&gt; "I’m testing my-pipeline_2", 
     "@timestamp" =&gt; 2020-02-05T22:20:43.250Z, 
       "@version" =&gt; "1" 
}
</pre><p>Nous voyons ci-dessus que la logique provenant de&nbsp;<code>02_filter.cfg</code> et<code>03_filter.cfg</code> est appliquée comme prévu.&nbsp;&nbsp;
</p> <strong><h3>Ordre d'exécution</h3></strong><p>Il faut savoir que Logstash ne tient pas compte de l'ordre des fichiers de l'expression glob. Il n'utilise l'expression glob que pour identifier les fichiers à inclure, puis les classe dans l'ordre alphabétique. Autrement dit, même si nous modifions la définition de <code>my-pipeline_2</code> pour que&nbsp;<code>03_filter.cfg</code> apparaisse avant&nbsp;<code>02_filter.cfg</code> dans l'expression glob, les événements passeront d'abord par le filtre défini dans&nbsp;<code>02_filter.cfg</code> avant de passer par celui de&nbsp;<code>03_filter.cfg</code>.
</p> <strong><h2>Pour conclure</h2></strong><p>L'utilisation d'expressions&nbsp;glob permet de créer des pipelines Logstash à partir de composants modulaires stockés en tant que fichiers distincts. Cela peut faciliter la gestion, la réutilisation et la lisibilité du code.&nbsp;
</p><p>Par ailleurs, je vous invite à consulter la page <a href="/guide/en/logstash/current/pipeline-to-pipeline.html">pipeline-to-pipeline communication</a>&nbsp;(communication de pipeline à pipeline), qui peut aussi vous aider à améliorer la modularité de votre implémentation Logstash.
</p>

blog-pipe-sky-banner.jpg

blog-pipe-sky-thumb.jpg

How to create maintainable and reusable Logstash pipelines

Comment créer des pipelines Logstash maintenables et réutilisables

Découvrez comment synchroniser les données de votre cluster Elasticsearch avec vos bases de données relationnelles existantes.

es-database-sync-banner.jpg

es-database-sync-thumb.jpg

How to keep Elasticsearch synchronized with a relational database using Logstash and JDBC

Synchroniser Elasticsearch avec une base de données relationnelle grâce à Logstash et JDBC

<p><a href="/fr/logstash">Logstash</a> est un pipeline open&nbsp;source côté serveur, destiné au traitement des données. Sa mission&nbsp;? Ingérer des données, les transformer, puis les envoyer vers une ou plusieurs sorties. Dans cet article de blog, je prendrai l'exemple du marché boursier pour vous montrer comment utiliser Logstash pour ingérer les données provenant de différentes bourses, puis envoyer les données correspondant à chacune d'entre elles vers une sortie distincte. Pour ce faire, nous suivrons les étapes suivantes&nbsp;:
</p>
<ol>
	<li>Créer des copies de chaque document depuis le flux d'entrée d'une place boursière.</li>
	<li>Filtrer chacune de ces copies, afin qu'elle ne contienne que les champs valides pour une place boursière donnée.</li>
	<li>Ajouter des métadonnées à chaque copie, afin d'indiquer à quelle place boursière correspondent les données qu'elle contient.</li>
	<li>Évaluer les métadonnées de chaque document, afin que ces derniers soient dirigés vers la bonne sortie.</li>
</ol>
<p>Remarque&nbsp;: dans cet article de blog, je n'utiliserai pas la <a href="/guide/en/logstash/6.5/pipeline-to-pipeline.html">communication pipeline-to-pipeline (disponible en bêta depuis la version&nbsp;6.5)</a>, qui pourrait aussi permettre d'atteindre certains des résultats décrits ici.
</p>
<h2>Exemple de fichier d'entrée</h2>
<p>En entrée vers Logstash, nous allons utiliser un fichier&nbsp;CSV contenant des valeurs boursières repères. Voici quelques exemples d'entrées CSV&nbsp;:
</p>
<pre class="prettyprint">1483230600,1628.75,1678.1,1772.8,2443.6
1483232400,1613.63,1688.5,1750.5,2460.2
1483234200,1606.51,1678.6,1718,2448.2
1483236000,1621.04,1684.1,1708.1,2470.4
</pre>
<p>Les valeurs séparées par une virgule correspondent aux valeurs temporelles du champ "time" et aux valeurs repères des places boursières suivantes&nbsp;: "DAX", "SMI", "CAC" et "FTSE" . Copiez et collez les lignes ci-dessus dans un fichier CSV que vous nommerez "stocks.csv", qui servira d'entrée dans l'exemple de pipeline Logstash.
</p>
<h2>Exemple de pipeline Logstash</h2>
<p>Ci-dessous, nous allons voir un pipeline Logstash qui exécute ce qui suit&nbsp;:
</p>
<ol>
	<li>Lire les valeurs boursières en entrée au format CSV depuis un fichier CSV.</li>
	<li>Mapper chaque ligne de l'entrée CSV vers un document&nbsp;JSON, dans lequel les colonnes&nbsp;CSV correspondent aux champs JSON suivants&nbsp;: &nbsp;"time", "DAX", "SMI", "CAC" et "FTSE".</li>
	<li>Convertir le champ "time" au format Unix.</li>
	<li>Utilisez le <a href="/guide/en/logstash/6.5/plugins-filters-clone.html">plug-in "clone filter"</a> pour créer deux copies de chaque document (outre le document d'origine). Le plug-in "clone filter" ajoute automatiquement un nouveau champ "type" à chaque nouvelle copie du document, où "type" correspond aux noms donnés dans le tableau de clones. Nous définissons les types pour qu'ils correspondent à "clone_for_SMI" ou "clone_for_FTSE", et chaque clone ne contiendra au final que les données correspondant à la bourse "SMI" ou à la bourse "FTSE". </li>
	<li>Pour chaque clone&nbsp;:</li>
	<ol>
		<li>Utilisez le <a href="/guide/en/logstash/6.5/plugins-filters-prune.html">plug-in "prune filter"</a> pour supprimer tous les champs, sauf ceux qui correspondent à la place boursière concernée.</li>
		<li>Ajoutez les <a href="/fr/blog/logstash-metadata">métadonnées</a> à chaque document correspondant au "type" ajouté par la fonction "clone". Cette opération est nécessaire, car nous utilisons la fonction "prune", qui supprime le "type" inséré par la fonction "clone", or, cette information est requise en sortie, afin de diriger le document vers la sortie appropriée.</li>
	</ol>
	<li>Utilisez le <a href="/guide/en/logstash/6.5/plugins-outputs-elasticsearch.html">plug-in de sortie Elasticsearch pour Logstash</a>, afin d'écrire les documents correspondant à chaque place de marché vers une sortie Elasticsearch&nbsp;distincte&nbsp;– la sortie étant déterminée par la valeur que nous avons définie dans le champ "metadata" (métadonnées) ajouté à l'étape&nbsp;5. Pour simplifier le code ci-dessous, chaque sortie Elasticsearch écrit vers un seul index compris dans un cluster Elasticsearch local. Si vous devez utiliser plusieurs clusters comme sorties, vous pouvez facilement modifier chaque déclaration de sortie Elasticsearch pour spécifier des <a href="/guide/en/logstash/6.5/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-hosts">hôtes Elasticsearch</a> uniques.</li>
</ol>
<p>Voici un pipeline Logstash qui exécute les étapes que nous venons de voir ci-dessus (j'y ai ajouté les numéros des étapes, ainsi que des commentaires).&nbsp;Copiez ce pipeline dans un fichier que vous nommerez "clones.conf"&nbsp;pour exécuter ce qui suit&nbsp;:
</p>
<pre class="prettyprint">## ÉTAPE 1
input {
  file {
    # n'oubliez pas de modifier le chemin, pour utiliser votre fichier stocks.csv
    path =&gt; "${HOME}/stocks.csv"
    # Les lignes ci-dessous assurent la relecture de toute l'entrée 
    # chaque fois que Logstash s'exécute (utile pour le débogage).
    start_position =&gt; "beginning"
    sincedb_path =&gt; "/dev/null"
  }
}
## ÉTAPE 2
filter {
   csv {
    columns =&gt; ["time","DAX","SMI","CAC","FTSE"]
    separator =&gt; ","
    convert =&gt; { 
      'DAX' =&gt; 'float'
      'SMI' =&gt; 'float'
      'CAC' =&gt; 'float'
      'FTSE' =&gt; 'float'
    }
  }
## ÉTAPE 3  
date {
    match =&gt; ['time', 'UNIX']
  }
## ÉTAPE 4
  # La ligne ci-dessous va créer 2 copies supplémentaire 
  # de chaque document (en plus de l'original, 
  # donc 3 documents au total). 
  # Un champ "type" est automatiquement ajouté à chaque copie 
  # Il correspond au nom donné dans le tableau.
  clone {
    clones =&gt; ['clone_for_SMI', 'clone_for_FTSE']
  }
## ÉTAPE 5
  if [type] == 'clone_for_SMI' {
    # Tout supprimer sauf "SMI"
    prune {
       whitelist_names =&gt; [ "SMI"]
    }
    mutate {
      add_field =&gt; { "[@metadata][type]" =&gt; "only_SMI" } 
    }
  } 
  else if [type] == 'clone_for_FTSE' {
    prune {
       whitelist_names =&gt; [ "FTSE"]
    }
    mutate {
      add_field =&gt; { "[@metadata][type]" =&gt; "only_FTSE" } 
    }
  } 
}
## ÉTAPE 6
output {
  # La sortie stdout ci-dessous sert uniquement au débogage 
  # Elle peut être supprimée
  stdout { 
    codec =&gt;  rubydebug {
      metadata =&gt; true
    }
  }
  if [@metadata][type] == 'only_SMI' {
    elasticsearch {
      index =&gt; "smi_data"
    }
  }
  else if [@metadata][type] == 'only_FTSE' {
    elasticsearch {
      index =&gt; "ftse_data"
    }
  }
  else {
    elasticsearch {
      index =&gt; "stocks_original"
    }
  }
}
</pre>
<h2>Tester le pipeline Logstash</h2>
<p>Pour tester ce pipeline avec les exemples de données CSV, vous pouvez exécuter la commande suivante en la modifiant, afin d'utiliser les chemins de votre système. Remarque&nbsp;: vous n'êtes pas obligé de spécifier "config.reload.automatic", mais cela vous permet de recharger automatiquement "clones.conf" sans devoir redémarrer Logstash&nbsp;:
</p>
<pre class="prettyprint">./logstash -f ./clones.conf --config.reload.automatic
</pre>
<p>Une fois que Logstash a lu le fichier "stocks.csv" et terminé le traitement, nous pouvons afficher les trois index ainsi créés, qui sont nommés "smi_data", "ftse_data" et "stocks_original".
</p>
<h2>Vérifier l'index SMI</h2>
<pre class="prettyprint">GET /smi_data/_search
</pre>
<p>Cela permet d'afficher les documents présentant la structure suivante. Vous remarquerez que seules les données "SMI" apparaissent dans l'index "smi_data".
</p>
<pre class="prettyprint">      {
        "_index": "smi_data",
        "_type": "doc",
        "_id": "_QRskWUBsYalOV9y9hGJ",
        "_score": 1,
        "_source": {
          "SMI": 1688.5    
        }
      }
</pre>
<h2>Vérifier l'index FTSE</h2>
<pre class="prettyprint">GET /ftse_data/_search
</pre>
<p><span style="background-color: initial;">Cela permet</span> d'afficher les documents présentant la structure suivante. Vous remarquerez que seules les données "FTSE" apparaissent dans l'index "ftse_data".
</p>
<pre class="prettyprint">      {
        "_index": "ftse_data",
        "_type": "doc",
        "_id": "AgRskWUBsYalOV9y9hL0",
        "_score": 1,
        "_source": {
          "FTSE": 2448.2
        }
      }
</pre>
<h2>Vérifier l'index des documents d'origine</h2>
<pre class="prettyprint">GET /stocks_originals/_search
</pre>
<p><span style="background-color: initial;">Cela permet</span> d'afficher les documents présentant la structure suivante. Vous remarquerez que la version d'origine non filtrée des documents apparaît dans l'index "stocks_original".
</p>
<pre class="prettyprint">      {
        "_index": "stocks_original",
        "_type": "doc",
        "_id": "-QRskWUBsYalOV9y9hFo",
        "_score": 1,
        "_source": {
          "host": "Alexanders-MBP",
          "@timestamp": "2017-01-01T00:30:00.000Z",
          "SMI": 1678.1,
          "@version": "1",
          "message": "1483230600,1628.75,1678.1,1772.8,2443.6",
          "CAC": 1772.8,
          "DAX": 1628.75,
          "time": "1483230600",
          "path": "/Users/arm/Documents/ES6.3/datasets/stocks_for_clones.csv",
          "FTSE": 2443.6
        }
      }
</pre>
<h2>Pour conclure</h2>
<p>Dans cet article de blog, nous avons étudié une infime partie des capacités de <a href="/fr/logstash">Logstash</a>. Nous avons pris l'exemple du marché boursier, pour montrer comment utiliser Logstash pour ingérer les données provenant de différentes places boursières, les traiter, puis les envoyer vers des sorties distinctes. Si vous utilisez la version d'essai de Logstash et de la <a href="/fr/products">Suite&nbsp;Elastic</a> et que vous avez des questions, n'hésitez pas à demander de l'aide sur nos <a href="http://discuss.elastic.co">forums de discussions publiques</a>.
</p>

log-split-route-banner.jpg

log-split-route-thumb.jpg

Using Logstash to Split Data and Send it to Multiple Outputs

Utiliser Logstash pour fractionner les données et les envoyer vers différentes sorties

Découvrez comment identifier les documents en double dans Elasticsearch et les supprimer, à l’aide de Logstash ou d’un script Python personnalisé.

<p>La plupart des systèmes qui intègrent des données dans Elasticsearch tirent parti des <a href="/guide/en/elasticsearch/reference/current/docs-index_.html#_automatic_id_generation">valeurs d’ID auto-générées d’Elasticsearch</a> lors de l’insertion de nouveaux documents. Le problème, c’est que si la source de données envoie le même document par inadvertance plusieurs fois à Elasticsearch et si une valeur auto-générée <code>_id</code> est utilisée pour chaque document qu’Elasticsearch insère, alors ce document sera stocké plusieurs fois dans Elasticsearch avec des valeurs <code>_id</code> différentes. Si une telle situation se produit, il peut être nécessaire de rechercher et de supprimer les documents en double. Dans cet article de blog, nous allons donc voir comment identifier les documents en double et les supprimer à partir d’Elasticsearch (1) avec Logstash, ou (2) avec un code personnalisé écrit en Python.
</p><h2>Exemple de structure de documents</h2><p>Dans le cadre de cet article de blog, nous allons partir du principe que les documents se trouvant dans le cluster Elasticsearch ont la structure suivante. Il s’agit ici d’un ensemble de données qui contient des documents représentant des transactions boursières.
</p><pre class="prettyprint">   {
      "_index": "stocks",
      "_type": "doc",
      "_id": "6fo3tmMB_ieLOlkwYclP",
      "_version": 1,
      "found": true,
      "_source": {
        "CAC": 1854.6,
        "host": "Alexanders-MBP",
        "SMI": 2061.7,
        "@timestamp": "2017-01-09T02:30:00.000Z",
        "FTSE": 2827.5,
        "DAX": 1527.06,
        "time": "1483929000",
        "message": "1483929000,1527.06,2061.7,1854.6,2827.5\r",
        "@version": "1"
      }
    }
</pre><p>À partir de cette structure, nous supposerons que si plusieurs documents ont les mêmes valeurs pour les champs <code>["CAC", "FTSE", "SMI"]</code>, c’est qu’ils sont des doublons les uns des autres.
</p><h2>Utilisation de Logstash pour la déduplication des documents Elasticsearch</h2><p>Vous pouvez utiliser Logstash pour identifier les doublons dans un index Elasticsearch, puis les supprimer. Pour savoir comment procéder, consultez <a href="/fr/blog/logstash-lessons-handling-duplicates">cet article de blog sur la gestion des doublons avec Logstash</a>. Nous allons voir dans cette section un exemple concret de la mise en application de cette approche.
</p><p>Dans l’exemple ci-dessous, j’ai programmé une configuration Logstash simple qui lit les documents à partir d’un index d’un cluster Elasticsearch, puis qui utilise le <a href="/guide/en/logstash/current/plugins-filters-fingerprint.html">filtre d’empreinte</a> pour calculer une valeur <code>_id</code> unique pour chaque document selon le hachage des champs <code>["CAC", "FTSE", "SMI"]</code>, et pour finir, qui réécrit chaque document dans un nouvel index sur ce même cluster Elasticsearch de sorte que tous les documents en double soient enregistrés avec le même <code>_id</code>, et de là, éliminés.
</p><p>De plus, avec quelques modifications mineures, le même filtre Logstash pourrait être également appliqué aux futurs documents écrits dans le nouvel index, pour faire en sorte que les doublons soient supprimés en temps quasi-réel. Il suffirait de changer la section d’entrée de l’exemple ci-dessous pour accepter les documents de votre source d’entrée en temps réel plutôt que de sortir les documents d’un index existant.
</p><p>Sachez que l’utilisation de valeurs <code>_id</code> personnalisées (c.-à-d. un <code>_id</code> qui n’est pas généré par Elasticsearch) <a href="/guide/en/elasticsearch/reference/master/tune-for-indexing-speed.html#_use_auto_generated_ids">aura une incidence sur les performances d’écriture</a> de vos opérations d’index.
</p><p>Il est aussi important de signaler que, selon l’algorithme de hachage utilisé, cette approche peut théoriquement aboutir à un certain nombre de <a href="https://en.wikipedia.org/wiki/Collision_(computer_science)">collisions de hachages</a> pour la valeur <code>_id</code>, ce qui entraînerait le mappage de deux documents non identiques au même <code>_id</code>, et de là, causerait la perte de l’un de ces documents. Pour la plupart des cas pratiques, la probabilité d’une collision de hachages reste très faible. Dans cet article de blog, nous ne verrons pas en détail les différentes fonctions de hachage, car ce n’est pas le but. Néanmoins, vous devez choisir avec soin la fonction de hachage utilisée dans le filtre d’empreinte, car elle aura un impact sur les performances d’ingestion ainsi que sur le nombre de collisions de hachage.
</p><p>Vous trouverez ci-dessous une configuration Logstash simple pour dédupliquer un index existant à l’aide du filtre d’empreinte.
</p><pre class="prettyprint">input {
  # Lecture de tous les documents à partir d’Elasticsearch 
  elasticsearch {
    hosts =&gt; "localhost"
    index =&gt; "stocks"
    query =&gt; '{ "sort": [ "_doc" ] }'
  }
}
# Ce filtre a été mis à jour le 18&nbsp;février&nbsp;2019
filter {
    fingerprint {
        key =&gt; "1234ABCD"
        method =&gt; "SHA256"
        source =&gt; ["CAC", "FTSE", "SMI"]
        target =&gt; "[@metadata][generated_id]"
    &nbsp;&nbsp;&nbsp;&nbsp;concatenate_sources =&gt; true # &lt;-- Nouvelle ligne ajoutée depuis la date de publication d’origine
    }
}
output {
    stdout { codec =&gt; dots }
    elasticsearch {
        index =&gt; "stocks_after_fingerprint"
        document_id =&gt; "%{[@metadata][generated_id]}"
    }
}
</pre><h2>Script Python personnalisé pour la déduplication de documents Elasticsearch</h2><h3>Une approche efficace en termes de mémoire</h3><p>Si vous n’utilisez pas Logstash pour la déduplication, vous pouvez alors recourir à un script Python personnalisé. Dans cette approche, nous calculons le hachage des champs <code>["CAC", "FTSE", "SMI"]</code> que nous avons défini pour identifier un document de façon unique. Nous utilisons ensuite ce hachage comme une clé dans un dictionnaire Python, dans lequel la valeur associée de chaque entrée de dictionnaire représentera un éventail des valeurs <code>_id</code> des documents mappés au même hachage.
</p><p>Si plusieurs documents ont le même hachage, les documents en double mappés au même hachage peuvent être supprimés. Cependant, si vous êtes préoccupé par le fait qu’il puisse y avoir des <a href="https://en.wikipedia.org/wiki/Collision_(computer_science)">collisions de hachage</a>, alors le contenu des documents mappés au même hachage peut être examiné pour vérifier si ces documents sont réellement identiques, et si tel est le cas, les doublons peuvent être alors supprimés.
</p><h3>Analyse de l’algorithme de détection</h3><p>Prenons un index de 50&nbsp;Go. Supposons qu’il contienne des documents d’une taille moyenne de 0,4&nbsp;Ko. Il y aurait alors 125&nbsp;millions de documents dans l’index. Le volume de mémoire requis pour y stocker les structures de données de déduplication lors de l’utilisation d’un hachage md5 de 128&nbsp;bits serait de l’ordre de 128&nbsp;bits x 125&nbsp;millions = 2&nbsp;Go de mémoire, plus les valeurs <code>_id</code> de 160&nbsp;bits, qui nécessiteraient 160&nbsp;bits x 125&nbsp;millions = 2,5&nbsp;Go de mémoire. Cet algorithme aurait donc besoin d’une RAM de l’ordre de 4,5&nbsp;Go pour conserver toutes les structures de données pertinentes en mémoire. Il est possible de réduire considérablement cette empreinte mémoire en appliquant l’approche que nous allons voir dans la section suivante.
</p><h3>Amélioration de l’algorithme</h3><p>Dans cette section, nous allons voir comment améliorer notre algorithme pour réduire l’utilisation de mémoire ainsi que pour supprimer les documents en double de façon continue.
</p><p>Si vous stockez des données temporelles et que vous savez que les documents en double sont générés uniquement sur une période limitée les uns par rapport aux autres, vous pouvez alors améliorer l’empreinte mémoire de cet algorithme en exécutant de façon répétée l’algorithme sur un sous-ensemble de documents de l’index, chaque sous-ensemble correspondant à une fenêtre temporelle différente. Par exemple, si vous avez recueilli des données sur plusieurs années, vous pourriez utiliser des <a href="/guide/en/elasticsearch/reference/current/query-dsl-range-query.html">demandes de recherche de plage</a> dans votre champ d’horodatage (à l’intérieur d’un <a href="/guide/en/elasticsearch/reference/current/query-filter-context.html">contexte de filtre</a> pour de meilleures performances) pour parcourir votre ensemble de données semaine par semaine. Pour cela, il faudrait que l’algorithme soit exécuté 52&nbsp;fois (une fois par semaine). Et dans ce cas, cette approche permettrait de réduire la "pire" empreinte mémoire d’un facteur de 52.
</p><p>Dans l’exemple ci-dessus, vous pouvez être préoccupé par le fait de ne pas identifier les documents en double qui s’étendent sur plusieurs semaines. Partons du principe que les documents en double ne peuvent pas être séparés de plus de deux heures. Ce qu’il vous faudrait faire dans ce cas serait d’inclure dans l’exécution de l’algorithme les documents qui chevauchent d’au maximum deux heures le dernier ensemble de documents analysés par l’exécution précédente de l’algorithme. Reprenons le cas d’une analyse hebdomadaire. Il vous faudrait alors faire une recherche de documents sur 170&nbsp;heures (1&nbsp;semaine + 2&nbsp;h) pour vous assurer qu’aucun doublon n’a été oublié.
</p><p>Si vous souhaitez éliminer périodiquement les doublons de vos index de façon continue, vous pouvez exécuter cet algorithme sur les documents récemment reçus. La même logique que ci-dessus s’applique&nbsp;: assurez-vous que les documents récemment reçus sont inclus dans l’analyse avec un temps de chevauchement suffisant par rapport aux documents légèrement plus anciens pour vérifier qu’aucun doublon n’est oublié par inadvertance.
</p><h3>Utilisation du code Python pour identifier les doublons</h3><p>Le code ci-dessous montre comment les documents peuvent être évalués avec efficacité pour voir s’ils sont identiques, et supprimés si nécessaire. Notez toutefois que pour éviter toute suppression accidentelle de documents, dans cet exemple, nous n’exécutons pas réellement une opération de suppression. Une fonctionnalité de ce type est néanmoins simple à mettre en place.
</p><p>Le code permettant de dédupliquer les documents à partir d’Elasticsearch est aussi <a href="https://github.com/alexander-marquardt/deduplicate-elasticsearch">proposé sur github</a>.
</p><pre class="prettyprint">#!/usr/local/bin/python3
import hashlib
from elasticsearch import Elasticsearch
es = Elasticsearch(["localhost:9200"])
dict_of_duplicate_docs = {}
# La ligne suivante définit les champs qui seront
# utilisés pour déterminer si un document est un doublon ou non
keys_to_include_in_hash = ["CAC", "FTSE", "SMI"]
# Traitement des documents renvoyés par la recherche actuelle ou le défilement
def populate_dict_of_duplicate_docs(hits):
    for item in hits:
        combined_key = ""
        for mykey in keys_to_include_in_hash:
            combined_key += str(item['_source'][mykey])
        _id = item["_id"]
        hashval = hashlib.md5(combined_key.encode('utf-8')).digest()
        # Si la valeur du hachage (hashval) est nouvelle, nous créerons une nouvelle clé
        # dans dict_of_duplicate_docs, qui se verra attribuer
        # une valeur d’un tableau vide.
        # Nous intégrerons ensuite immédiatement la valeur _id dans le tableau.
        # Si la valeur hashval existe déjà, alors
        # nous indiquerons tout simplement la nouvelle valeur _id dans le tableau existant.
        dict_of_duplicate_docs.setdefault(hashval, []).append(_id)
# Créez une boucle avec tous les documents de l’index, puis renseignez la
# structure de données dict_of_duplicate_docs.
def scroll_over_all_docs():
    data = es.search(index="stocks", scroll='1m',  body={"query": {"match_all": {}}})
    # Obtenez l’ID de défilement
    sid = data['_scroll_id']
    scroll_size = len(data['hits']['hits'])
    # Avant de procéder au défilement, traitez le lot de résultats en cours
    populate_dict_of_duplicate_docs(data['hits']['hits'])
    while scroll_size &gt; 0:
        data = es.scroll(scroll_id=sid, scroll='2m')
        # Traitez le lot de résultats en cours
        populate_dict_of_duplicate_docs(data['hits']['hits'])
        # Mettez l’ID de défilement à jour
        sid = data['_scroll_id']
        # Obtenez le nombre de résultats renvoyés lors du dernier défilement
        scroll_size = len(data['hits']['hits'])
def loop_over_hashes_and_remove_duplicates():
    # Parcourez les valeurs de hachage pour déterminer s’il existe
    # des hachages en double
    for hashval, array_of_ids in dict_of_duplicate_docs.items():
      if len(array_of_ids) &gt; 1:
        print("********** Duplicate docs hash=%s **********" % hashval)
        # Obtenez les documents mappés à la valeur hashval actuelle
        matching_docs = es.mget(index="stocks", doc_type="doc", body={"ids": array_of_ids})
        for doc in matching_docs['docs']:
            # Dans cet exemple, nous avons simplement imprimé les documents en double.
            # Il est tout à fait possible (et simple&nbsp;!) de modifier ce code pour supprimer les doublons
            # ici au lieu de les imprimer.
            print("doc=%s\n" % doc)
def main():
    scroll_over_all_docs()
    loop_over_hashes_and_remove_duplicates()
main()
</pre><h2>Conclusion</h2><p>Dans cet article de blog, nous avons vu deux méthodes pour la déduplication de documents dans Elasticsearch. La première méthode s’appuie sur Logstash pour supprimer les documents en double. La deuxième consiste à utiliser un script Python personnalisé pour identifier et supprimer les doublons.
</p><p>Pour toute question concernant la déduplication des documents Elasticsearch, ou pour tout autre sujet concernant Elasticsearch, jetez un œil à nos <a href="https://discuss.elastic.co/">forums de discussion</a> pour obtenir des informations et des perspectives intéressantes.
</p>

Articles by Alex Marquardt

Comment créer des pipelines Logstash maintenables et réutilisables

Synchroniser Elasticsearch avec une base de données relationnelle grâce à Logstash et JDBC

Utiliser Logstash pour fractionner les données et les envoyer vers différentes sorties

Rechercher et supprimer des documents en double dans Elasticsearch

Produits

Entreprise

Ressources