Ingest Pipelines: Datentransformation vor dem Indexieren

1. Wo Datentransformation stattfinden sollte: Anwendung oder Index

Jede Datenquelle liefert Rohdaten in einem Format, das selten exakt dem Zielschema des Suchindex entspricht. Preise kommen als Strings mit Tausendertrennzeichen, Kategoriezuordnungen als kommagetrennte Freitextfelder, Beschreibungen als unstrukturierter Fließtext mit eingebetteten technischen Kennzahlen. Die klassische Antwort lautet, diese Aufbereitung im Anwendungscode zu erledigen, bevor ein Dokument überhaupt an Elasticsearch gesendet wird, etwa in einem Magento-Indexer oder einem separaten Export-Skript, das die Produktdaten aus der Datenbank liest und in das Zielschema überführt.

Diese anwendungsseitige Lösung ist nicht falsch, verlagert die Logik aber vollständig in eigenen Code, der gepflegt, getestet und bei jeder Schemaänderung angepasst werden muss. Ingest Pipelines bieten eine Alternative direkt in Elasticsearch: Die Transformationslogik liegt deklarativ als Pipeline-Definition im Cluster, lässt sich unabhängig von der sendenden Anwendung ändern und wirkt automatisch auf jedes Dokument, das über diese Pipeline läuft, unabhängig davon, aus welcher Quelle es stammt.

2. Ingest-Pipelines: Aufbau aus einer geordneten Prozessorkette

Eine Ingest Pipeline ist eine benannte, über die Ingest API definierte Abfolge von Prozessoren, die ein Dokument nacheinander durchläuft, bevor es tatsächlich in einen Shard geschrieben wird. Jeder Prozessor erhält das Dokument im aktuellen Zwischenzustand, verändert Felder, fügt neue hinzu oder entfernt vorhandene, und reicht das Ergebnis an den nächsten Prozessor weiter. Diese Verarbeitung läuft auf dem Knoten, der den Schreibvorgang koordiniert, bevor die eigentliche Indexierungslogik greift.

Referenziert wird eine Pipeline entweder explizit über den Parameter pipeline bei einem Index- oder Bulk-Request, oder implizit über die Indexeinstellung index.default_pipeline, die automatisch für jedes Dokument dieses Index greift. Zusätzlich existiert index.final_pipeline, die stets als letzte Pipeline läuft, unabhängig davon, welche Pipeline zuvor explizit angegeben wurde, etwa um unabhängig von der Quelle immer einen Zeitstempel oder eine Versionskennung zu setzen.

PUT _ingest/pipeline/product-basic-cleanup
{
  "description": "Trim, lowercase and default currency for product docs",
  "processors": [
    { "trim": { "field": "sku" } },
    { "lowercase": { "field": "sku" } },
    { "set": { "field": "currency", "value": "EUR", "override": false } }
  ]
}

3. grok-Prozessor: unstrukturierte Strings in benannte Felder zerlegen

Der grok-Prozessor parst unstrukturierten Text anhand vordefinierter und eigener Muster in benannte Felder, ein Konzept, das viele aus Logstash kennen. Ein Muster wie %{IP:client_ip} %{WORD:method} %{URIPATHPARAM:request} zerlegt eine Logzeile in einzelne, durchsuchbare und aggregierbare Felder, statt sie als einen einzigen unstrukturierten String zu speichern. Elasticsearch bringt eine umfangreiche Bibliothek fertiger Muster für gängige Formate wie Apache-Logs, Syslog oder IP-Adressen mit, eigene Muster lassen sich ergänzen.

Im Produktkontext eignet sich grok für halbstrukturierte Herstellerfelder, etwa eine technische Kennzeichnung wie Art.-Nr. 4711-XL-BLAU, aus der sich Artikelnummer, Größe und Farbe als eigenständige, filterbare Felder extrahieren lassen. Scheitert das Muster an einem Dokument, weil das Format von der Erwartung abweicht, markiert grok das Dokument standardmäßig mit einem Tag im Feld _ingest._value.tags, statt den gesamten Schreibvorgang fehlschlagen zu lassen.

{
  "grok": {
    "field": "raw_article_code",
    "patterns": ["Art\\.-Nr\\. %{DATA:article_number}-%{DATA:size}-%{WORD:color}"],
    "tag": "parse-article-code"
  }
}

4. script-Prozessor: Painless für individuelle Transformationslogik

Wo vorgefertigte Prozessoren wie convert, split oder rename nicht ausreichen, übernimmt der script-Prozessor beliebige Painless-Logik direkt innerhalb der Pipeline. Das eignet sich für Fälle, die sich nicht mit einer einzelnen deklarativen Anweisung lösen lassen, etwa eine bedingte Preisberechnung abhängig von mehreren Feldern gleichzeitig, oder eine Normalisierung, die je nach Herkunftssystem unterschiedliche Regeln anwendet.

Wichtig ist, den script-Prozessor gezielt und sparsam einzusetzen, da jede Codezeile bei jedem Dokument erneut ausgeführt wird und damit direkt in den Schreibdurchsatz eingeht. Für einfache Umbenennungen, Typkonvertierungen oder das Setzen von Standardwerten existieren dedizierte, deutlich günstigere Prozessoren, die dieselbe Aufgabe ohne den Overhead der Painless-Auswertung erledigen.

{
  "script": {
    "source": "ctx.price_normalized = Double.parseDouble(ctx.price_raw.replace(',', '.'))"
  }
}

5. enrich-Prozessor: Referenzdaten aus einem anderen Index nachladen

Der enrich-Prozessor führt beim Schreiben eines Dokuments einen Nachschlagevorgang gegen einen vorbereiteten Referenzbestand durch und fügt passende Felder hinzu, vergleichbar mit einem Left Join zur Indexierungszeit. Voraussetzung ist eine Enrich Policy, die festlegt, welcher Quellindex als Referenz dient, über welches Match-Feld der Abgleich erfolgt und welche Zielfelder übernommen werden. Nach dem Ausführen der Policy erzeugt Elasticsearch daraus einen optimierten, versteckten Enrich-Index.

Typisches Beispiel ist eine Herstellerreferenz: Ein Produktdokument enthält nur eine Marken-ID, der enrich-Prozessor lädt daraus zur Indexierungszeit den vollständigen Markennamen, das Herkunftsland und ein Logo-Kürzel und schreibt diese Felder direkt in das Produktdokument. Damit muss die suchende Anwendung diese Zusatzdaten nicht mehr per separatem Aufruf nachladen, sie liegen bereits denormalisiert im durchsuchbaren Dokument.

PUT _enrich/policy/brand-lookup
{
  "match": {
    "indices": "brand-reference",
    "match_field": "brand_id",
    "enrich_fields": ["brand_name", "brand_country", "brand_logo_code"]
  }
}
POST _enrich/policy/brand-lookup/_execute

{
  "enrich": {
    "policy_name": "brand-lookup",
    "field": "brand_id",
    "target_field": "brand"
  }
}

6. Praxisbeispiel: eine vollständige Pipeline zur Produktdaten-Normalisierung

In der Praxis kombiniert eine einzelne Produktpipeline meist mehrere Prozessoren: trim und lowercase normalisieren die SKU, convert wandelt einen als String gelieferten Preis in einen numerischen Typ um, split zerlegt eine kommagetrennte Kategorienliste in ein Array, und der enrich-Prozessor ergänzt Markendaten aus der Referenztabelle. Jeder Prozessor löst dabei genau ein klar abgegrenztes Problem, statt eine einzelne, schwer wartbare Skriptlogik für alles zuständig zu machen.

Diese Aufteilung in einzelne, benannte Prozessoren macht die Pipeline auch für Kollegen nachvollziehbar, die sie nicht selbst geschrieben haben, und erlaubt es, einzelne Schritte gezielt zu testen oder auszutauschen, ohne die gesamte Transformationslogik neu zu schreiben. Für einen Magento-Produktexport bedeutet das: Die Exportlogik in der Anwendung liefert möglichst nah am Rohformat der Datenbank, die Pipeline übernimmt die eigentliche Normalisierung auf das Suchschema.

PUT _ingest/pipeline/product-normalize
{
  "processors": [
    { "trim": { "field": "sku" } },
    { "lowercase": { "field": "sku" } },
    { "convert": { "field": "price_raw", "type": "float", "target_field": "price" } },
    { "split": { "field": "categories_raw", "separator": "," } },
    { "enrich": { "policy_name": "brand-lookup", "field": "brand_id", "target_field": "brand" } }
  ]
}

7. Fehlerbehandlung: on_failure und ignore_failure gezielt einsetzen

Scheitert ein Prozessor an einem einzelnen Dokument, etwa weil ein erwartetes Feld fehlt oder ein Konvertierungsversuch nicht aufgeht, bricht ohne weitere Konfiguration die gesamte Pipeline für dieses Dokument ab und der Schreibvorgang schlägt fehl. Mit ignore_failure: true auf Prozessorebene lässt sich ein einzelner, unkritischer Prozessor übergehen, während die restliche Pipeline normal weiterläuft, sinnvoll etwa für einen enrich-Schritt, dessen Fehlen das Dokument nicht unbrauchbar macht.

Für kritischere Fälle definiert on_failure, sowohl auf Prozessor- als auch auf Pipeline-Ebene, eine alternative Prozessorkette, die im Fehlerfall greift. Eine bewährte Praxis ist, im on_failure-Block das Dokument mit den Fehlerdetails aus _ingest.on_failure_message zu markieren und in einen separaten Fehlerindex umzuleiten, statt es stillschweigend zu verwerfen. So bleiben fehlerhafte Rohdaten sichtbar und lassen sich gezielt nachbearbeiten.

8. Pipelines gefahrlos testen mit der Simulate API

Bevor eine Pipeline produktiv auf echte Schreibvorgänge angewendet wird, lässt sie sich über POST _ingest/pipeline/_simulate gegen Beispieldokumente testen, ohne dass tatsächlich etwas im Index landet. Übergeben werden entweder die vollständige Pipeline-Definition inline oder der Name einer bereits gespeicherten Pipeline zusammen mit einer Liste von Testdokumenten, die die erwarteten Rohdatenformate abdecken sollten, einschließlich bewusst fehlerhafter Beispiele.

Mit dem Parameter verbose=true zeigt die Antwort nicht nur das Endergebnis, sondern den Dokumentzustand nach jedem einzelnen Prozessor, was besonders bei mehrstufigen Pipelines beim Debugging hilft. Ein typischer Arbeitsablauf besteht darin, neue oder geänderte Prozessoren zunächst ausschließlich über die Simulate API gegen eine Sammlung repräsentativer Testdokumente zu prüfen, bevor die Pipeline überhaupt mit einem echten Index verknüpft wird.

POST _ingest/pipeline/product-normalize/_simulate?verbose=true
{
  "docs": [
    { "_source": { "sku": " ABC-123 ", "price_raw": "19,90", "categories_raw": "Schuhe,Sale" } }
  ]
}

9. Performance-Implikationen: wann die Anwendungsseite die bessere Wahl bleibt

Ingest-Verarbeitung läuft synchron auf dem Knoten, der den Schreibvorgang entgegennimmt, und geht damit direkt in dessen CPU-Budget und in die Latenz des Bulk-Requests ein. Ein aufwendiger grok-Ausdruck mit vielen Alternativen oder ein rechenintensiver script-Prozessor, der auf jedes einzelne Dokument angewendet wird, kann bei hohem Indexierungsdurchsatz zu einem spürbaren Engpass werden, gerade während einer initialen Massenindexierung eines großen Katalogs.

Als Faustregel eignen sich Ingest Pipelines besonders für strukturelle Normalisierung nahe am Zielschema, für einfache Feldumbenennungen, Typkonvertierungen und Anreicherung aus stabilen Referenzdaten. Rechenintensive, geschäftslogiklastige Transformationen, die ohnehin bereits im Anwendungscode existieren oder komplexe externe Abhängigkeiten haben, bleiben oft besser auf der Anwendungsseite aufgehoben, wo sie sich unabhängig vom Indexierungsdurchsatz skalieren und getrennt testen lassen.

10. Zusammenfassung

11. FAQ: Ingest Pipelines: Das Wichtigste auf einen Blick