36 Big Picture: Zusammenspiel von Ingest-Pipelines

Ingest-Pipelines sind ein zentraler Bestandteil der Datenverarbeitung in OpenSearch. Um ihr volles Potenzial zu verstehen, ist es wichtig, ihr Zusammenspiel mit anderen Komponenten und Prozessen in einem realistischen Kontext zu betrachten. Dieses Kapitel gibt eine Übersicht über das Big Picture der Ingest-Pipelines und beschreibt ihren Platz im Datenfluss von der Datenerfassung bis zur Analyse.

36.1 Datenfluss: Von der Quelle zum Index

Der typische Datenfluss in OpenSearch lässt sich in folgende Schritte unterteilen:

1. Datenerfassung: Daten stammen aus verschiedenen Quellen wie Log-Dateien, APIs, Datenbanken oder IoT-Geräten.
2. Datenanlieferung: Mittels Tools wie Logstash, Beats oder direkter API-Requests werden die Daten an OpenSearch übermittelt.
3. Ingest-Pipelines: Die Rohdaten durchlaufen eine oder mehrere Pipelines, um sie zu transformieren, anzureichern oder zu bereinigen.
4. Indexierung: Die verarbeiteten Daten werden in den entsprechenden Index geschrieben.
5. Analyse und Abfrage: Benutzer oder Anwendungen greifen über Queries auf die indizierten Daten zu und nutzen sie für Dashboards, Berichte oder andere Zwecke.

36.2 Rolle der Ingest-Pipelines im Datenfluss

Die Ingest-Pipeline ist der zentrale Verarbeitungsschritt zwischen der Datenerfassung und der Indexierung. Sie stellt sicher, dass:

• Daten bereinigt werden (z. B. Entfernung von Leerzeichen, Konsistenz in Formaten).
• Daten angereichert werden (z. B. Hinzufügen von Metadaten wie Timestamps oder Kategorien).
• Fehler korrigiert werden (z. B. Umwandlung von Datentypen oder Standardisierung von Feldern).

36.3 Exemplarisches Big Picture-Beispiel

36.3.1 Szenario: Log-Verarbeitung für ein Online-Shop-System

Ein Online-Shop möchte Log-Daten aus seinen Webservern analysieren, um Einblicke in Nutzerverhalten und Performance zu gewinnen.

1. Datenerfassung: Logs werden in Echtzeit von den Webservern generiert und durch Filebeat gesammelt.

2. Datenanlieferung: Filebeat sendet die Log-Daten über einen HTTP-Request an OpenSearch und gibt dabei die zu verwendende Ingest-Pipeline an.

3. Verarbeitung in der Ingest-Pipeline:

   • Bereinigung: Entfernen von redundanten Leerzeichen und Sonderzeichen.
   • Anreicherung: Hinzufügen von Feldern wie processing_timestamp und Geo-Informationen basierend auf IP-Adressen.
   • Standardisierung: Konvertieren von Datumsangaben in ein einheitliches ISO-Format.
   • Validierung: Prüfen, ob erforderliche Felder wie status_code oder response_time vorhanden und gültig sind.

   Beispiel einer Pipeline für die Log-Verarbeitung:

   PUT _ingest/pipeline/log_processing_pipeline
   {
     "description": "Pipeline zur Verarbeitung von Webserver-Logs",
     "processors": [
       {
         "gsub": {
           "field": "message",
           "pattern": "\\s+",
           "replacement": " "
         }
       },
       {
         "geoip": {
           "field": "client_ip",
           "target_field": "geo_location"
         }
       },
       {
         "date": {
           "field": "timestamp",
           "formats": ["dd/MMM/yyyy:HH:mm:ss Z"],
           "target_field": "@timestamp"
         }
       },
       {
         "set": {
           "field": "processing_timestamp",
           "value": "{{_ingest.timestamp}}"
         }
       }
     ]
   }

4. Indexierung: Die verarbeiteten Log-Daten werden im Index webserver-logs gespeichert.

5. Analyse und Abfrage:

   • Dashboards in Kibana oder Grafana zeigen Nutzungsstatistiken, geografische Verteilung der Zugriffe und durchschnittliche Antwortzeiten an.
   • Alerts informieren bei anomalen Mustern wie einer hohen Anzahl von Fehlerstatuscodes.

36.4 Vorteile des Einsatzes von Ingest-Pipelines

• Automatisierung: Wiederkehrende Transformationen werden automatisch und konsistent durchgeführt.
• Flexibilität: Pipelines können einfach angepasst werden, um auf Änderungen in den Datenquellen zu reagieren.
• Leistungssteigerung: Vorverarbeitete Daten ermöglichen effizientere Queries und Analysen.

36.5 Beispiel

Hier ist ein vollständiges Beispiel, das eine Ingest-Pipeline definiert, Daten durch diese Pipeline verarbeitet und die Daten in einen Index schreibt.

36.5.1 Ziel: Verarbeitung von Produktdaten

• Pipeline-Aufgabe:
  • Entferne Leerzeichen.
  • Konvertiere den Produktnamen in Kleinbuchstaben.
  • Füge das aktuelle Verarbeitungsdatum hinzu.
• Daten: Produktinformationen.
• Ergebnis: Verarbeitete Daten werden in den Index products geschrieben.

36.5.2 Ingest-Pipeline erstellen

PUT _ingest/pipeline/product_pipeline
{
  "description": "Pipeline zur Verarbeitung von Produktdaten",
  "processors": [
    {
      "trim": {
        "field": "product_name"
      }
    },
    {
      "lowercase": {
        "field": "product_name"
      }
    },
    {
      "set": {
        "field": "processing_date",
        "value": "{{_ingest.timestamp}}"
      }
    }
  ]
}

• Beschreibung der Schritte:
  • trim: Entfernt Leerzeichen aus dem Feld product_name.
  • lowercase: Wandelt den Inhalt von product_name in Kleinbuchstaben um.
  • set: Fügt ein neues Feld processing_date mit dem aktuellen Zeitstempel hinzu.

36.5.3 Dokument durch die Pipeline verarbeiten und in den Index schreiben

PUT /products/_doc/1?pipeline=product_pipeline
{
  "product_name": "  OpenSearch T-Shirt ",
  "price": 19.99,
  "stock": 100
}

• Parameter ?pipeline=product_pipeline: Der Pipeline wird die Verarbeitung des Dokuments vor dem Schreiben in den Index übertragen.

36.5.4 Ergebnis prüfen

Rufe das verarbeitete Dokument aus dem Index ab:

GET /products/_doc/1

• Erwartetes Ergebnis:

{
  "_index": "products",
  "_id": "1",
  "_source": {
    "product_name": "opensearch t-shirt",
    "price": 19.99,
    "stock": 100,
    "processing_date": "2025-01-14T12:00:00.000Z"
  }
}

36.5.5 Erklärung des Ablaufs

1. Das ursprüngliche Dokument enthält den Produktnamen mit führenden und nachfolgenden Leerzeichen sowie Großbuchstaben.
2. Die Ingest-Pipeline:
   • Entfernt die Leerzeichen.
   • Wandelt den Produktnamen in Kleinbuchstaben um.
   • Fügt das Feld processing_date hinzu.
3. Das verarbeitete Dokument wird im Index products gespeichert.

Mit diesem Ansatz können Sie Schritt für Schritt nachvollziehen, wie Ingest-Pipelines in OpenSearch Daten modifizieren und sie in einem Index ablegen. Soll ich ein weiteres Beispiel mit anderen Prozessoren erstellen?