Gestion efficace des données avec la commande API _bulk d'Elasticsearch

L'API _bulk d'Elasticsearch est l'outil approprié lorsque votre application doit indexer, mettre à jour ou supprimer de nombreux documents sans payer un coût de requête HTTP par document. Ce qui pose problème, c'est le corps de la requête : il s'agit de JSON délimité par des sauts de ligne, et non d'un tableau JSON joliment formaté.

Utilisez _bulk lorsque vous chargez des logs, synchronisez des enregistrements depuis une autre base de données, ou appliquez un lot de suppressions de nettoyage. Vous devez toujours inspecter chaque élément de la réponse, car une opération peut échouer alors que la requête HTTP globale réussit.

Comprendre la structure de l'API _bulk

L'API _bulk accepte du JSON délimité par des sauts de ligne, généralement appelé NDJSON. Chaque action est définie sur une ligne. Les actions qui nécessitent un corps de document utilisent la ligne suivante comme source ou charge utile de mise à jour. La dernière ligne doit également se terminer par un saut de ligne.

Composants clés d'une requête _bulk :

Ligne d'action et de métadonnées : Cette ligne spécifie le type d'opération (index, create, update ou delete), l'index cible et éventuellement l'ID du document. Les types de document ne sont pas utilisés dans les API Elasticsearch modernes.
Ligne source : Cette ligne contient le document JSON réel à indexer ou à mettre à jour. Cette ligne est omise pour les opérations delete.
Délimiteur de saut de ligne : Chaque paire action/métadonnées et sa source correspondante (le cas échéant) doivent être séparées par un caractère de saut de ligne (\n). Le corps entier de la requête doit se terminer par un saut de ligne.

Exemple de structure :

{ "index": { "_index": "my-index", "_id": "1" } }
{ "field1": "value1" }
{ "delete": { "_index": "my-index", "_id": "2" } }

Ou pour une opération de suppression :

curl -sS -H 'Content-Type: application/x-ndjson' \
  -X POST 'http://localhost:9200/_bulk' \
  --data-binary @bulk.ndjson

Effectuer des opérations courantes avec _bulk

L'API _bulk est polyvalente et peut gérer un mélange d'opérations dans une seule requête. C'est là que réside sa véritable puissance, vous permettant d'effectuer des manipulations de données complexes en un seul aller-retour.

Indexation de plusieurs documents

Pour indexer plusieurs documents, vous utilisez l'action index. Si un document avec l'ID spécifié existe déjà, index l'écrasera. Si vous voulez vous assurer qu'un document n'est indexé que s'il n'existe pas déjà, utilisez plutôt l'action create.

Exemple : Indexation de deux nouveaux documents.

{ "index": { "_index": "my-index", "_id": "1" } }
{ "field1": "value1", "field2": "value2" }
{ "index": { "_index": "my-index", "_id": "2" } }
{ "field1": "another_value", "field2": "different_value" }

Mise à jour de documents

La mise à jour de documents peut être effectuée à l'aide de l'action update. Vous spécifiez l'ID du document à mettre à jour et fournissez un document partiel avec les champs que vous souhaitez modifier. Si vous souhaitez utiliser un script pour la mise à jour, vous pouvez le faire dans l'action update.

Exemple : Mise à jour d'un champ dans un document existant.

{ "update": { "_index": "my-index", "_id": "1" } }
{ "doc": { "field1": "updated_value" } }

Suppression de documents

Pour supprimer des documents, vous utilisez l'action delete, en spécifiant _index et _id du document à supprimer. Aucun document source n'est requis pour les opérations de suppression.

Exemple : Suppression d'un document.

{ "delete": { "_index": "my-index", "_id": "2" } }

Combinaison d'opérations

L'efficacité réelle vient du mélange de ces opérations. Vous pouvez indexer de nouveaux documents, mettre à jour des documents existants et en supprimer d'autres dans la même requête _bulk.

Exemple : Indexation, mise à jour et suppression dans une seule requête.

{ "index": { "_index": "my-index", "_id": "3" } }
{ "field1": "new_document_field", "field2": "new_document_value" }
{ "update": { "_index": "my-index", "_id": "1" } }
{ "doc": { "field1": "further_updated_value" } }
{ "delete": { "_index": "my-index", "_id": "2" } }

Gestion des réponses

L'API _bulk renvoie une réponse JSON qui détaille le résultat de chaque opération individuelle. Il est crucial d'analyser cette réponse pour vérifier que toutes les opérations ont réussi et pour identifier les éventuelles erreurs.

La réponse contiendra un tableau items, où chaque élément correspond à l'une des opérations de votre requête, dans le même ordre. Chaque élément inclura l'opération index, create, update ou delete, ainsi que son statut (par exemple, created, updated, deleted, noop) et d'autres métadonnées pertinentes.

Exemple d'extrait de réponse :

{
  "took": 150,
  "errors": false,
  "items": [
    {
      "index": {
        "_index": "my-index",
        "_id": "3",
        "version": 1,
        "result": "created",
        "_shards": {"total": 2, "successful": 1, "failed": 0},
        "_seq_no": 0,
        "_primary_term": 1
      }
    },
    {
      "update": {
        "_index": "my-index",
        "_id": "1",
        "version": 2,
        "result": "updated",
        "_shards": {"total": 2, "successful": 1, "failed": 0},
        "_seq_no": 1,
        "_primary_term": 1
      }
    },
    {
      "delete": {
        "_index": "my-index",
        "_id": "2",
        "version": 2,
        "result": "deleted",
        "_shards": {"total": 2, "successful": 1, "failed": 0},
        "_seq_no": 2,
        "_primary_term": 1
      }
    }
  ]
}

Si une opération échoue, le champ errors de premier niveau dans la réponse sera true, et l'élément individuel de l'opération ayant échoué contiendra un objet error détaillant le problème.

Bonnes pratiques et conseils

Taille du lot : Des lots très volumineux peuvent solliciter la mémoire du client, les nœuds coordinateurs et les nœuds de données. Commencez par des charges utiles modestes, mesurez le débit et les taux de rejet, puis ajustez en fonction de votre cluster et de la taille des documents.
Gestion des erreurs : Analysez toujours la réponse pour détecter les erreurs. Implémentez une logique de nouvelle tentative pour les erreurs transitoires si nécessaire.
Délimiteurs de saut de ligne : Assurez-vous que les caractères de saut de ligne (\n) sont correctement utilisés entre chaque objet JSON. Un formatage incorrect est une cause fréquente d'échec de l'API _bulk.
Parallélisation : Pour des taux d'ingestion très élevés, envisagez d'envoyer plusieurs requêtes _bulk en parallèle, mais soyez conscient de la capacité de votre cluster.
create vs. index : Utilisez create lorsque vous souhaitez que l'opération échoue si l'ID existe déjà. Utilisez index lorsque le remplacement d'un document existant est acceptable.
Clients API : La plupart des bibliothèques clientes Elasticsearch fournissent des méthodes pratiques pour construire et exécuter des requêtes _bulk, en abstraisant une partie du formatage manuel.

En résumé

L'API _bulk est rapide car elle réduit la surcharge des requêtes, mais elle n'est sûre que lorsque votre client traite la réponse comme une liste de résultats individuels. Envoyez du NDJSON valide avec Content-Type: application/x-ndjson, maintenez les lots à une taille que votre cluster peut absorber, et ne réessayez que les opérations qui ont échoué pour des raisons transitoires.