Pipes and Filters und Pipeline-Architekturen

Definition und Ursprung

Im Stil Pipes and Filters ist die Verarbeitung eine Kette unabhängiger Schritte. Jeder Filter liest Daten von seiner Eingabe, erledigt eine Aufgabe damit und schreibt das Ergebnis in seine Ausgabe. Jede Pipe transportiert Daten in eine Richtung von einem Filter zum nächsten, meist mit einem Puffer dazwischen. Die Beschreibung von Garlan und Shaw (1994) nennt die Einschränkungen, die dem Stil seine Eigenschaften geben: Filter müssen unabhängig sein, dürfen keinen Zustand mit anderen Filtern teilen und dürfen die Identität der Filter an beiden Enden ihrer Pipes nicht kennen. Ein Filter kennt nur das Format dessen, was er liest und was er schreibt.

Die bekannteste Heimat des Stils ist Unix. In einem Memo vom 11. Oktober 1964 schrieb Doug McIlroy, es solle Möglichkeiten geben, Programme wie einen Gartenschlauch zu koppeln und ein weiteres Segment anzuschrauben, wenn die Daten auf andere Weise bearbeitet werden müssen. Ken Thompson implementierte 1973 in Version 3 Unix den Systemaufruf pipe und Pipes in der Shell, und die Notation | folgte in Version 4. Die daraus entstandene Unix-Philosophie, Programme zu schreiben, die eine Sache gut machen und über eine universelle Textschnittstelle zusammenarbeiten, ist der Stil Pipes and Filters, angewandt auf den Werkzeugsatz eines ganzen Betriebssystems. POSA (1996) katalogisierte Pipes and Filters als Architekturmuster, und Hohpe und Woolfs Enterprise Integration Patterns (2003) wandten es auf das Messaging zwischen Systemen an.

Richards und Ford nennen denselben Stil Pipeline-Architektur und unterscheiden vier Arten von Filtern. Ein Producer (oder Quelle) startet den Fluss und liest aus einer Datei, einer Datenbank oder einer Queue. Ein Transformer verändert jeden Datensatz: parsen, umwandeln, anreichern. Ein Tester prüft eine Bedingung und gibt den Datensatz weiter oder verwirft ihn, manchmal leitet er ihn an eine von mehreren Ausgaben. Ein Consumer (oder Senke) beendet den Fluss und schreibt in eine Datenbank, eine Datei oder auf einen Bildschirm.

So funktioniert es

Die Einschränkungen erzeugen die Vorteile unmittelbar. Weil ein Filter keinen Zustand teilt und seine Nachbarn nicht kennt, lässt er sich für sich schreiben und testen, mit einer Liste von Eingabedatensätzen und einer erwarteten Liste von Ausgaben; er lässt sich in anderen Pipelines wiederverwenden; und Filter lassen sich hinzufügen, entfernen oder umordnen, indem man nur ändert, wie die Pipeline zusammengesetzt ist. Weil Datensätze inkrementell fließen, kann ein Filter mit dem ersten Datensatz beginnen, bevor der vorherige Filter den letzten fertig hat; so laufen die Stufen nebenläufig, und die ganze Pipeline braucht Speicher für nur wenige Datensätze auf einmal, nicht für den gesamten Datenbestand.

Die Pipe verbirgt, wie Daten sich bewegen. In einer Unix-Shell ist sie ein Puffer des Betriebssystems zwischen zwei Prozessen; unter Linux fasst er standardmäßig 64 KiB, und ein Schreiber, der ihn füllt, blockiert einfach, bis der Leser aufholt. In einem Python-Programm kann sie ein Generator sein, sodass jeder Filter den nächsten Datensatz vom vorherigen abruft. In einer Datenplattform kann sie eine Message Queue oder ein Topic in einem Log wie Apache Kafka sein, wodurch Filter auf verschiedenen Maschinen laufen und Neustarts überstehen können. Der Filtercode ändert sich dabei kaum; die betrieblichen Eigenschaften ändern sich sehr stark.

Es gibt Varianten der strengen Form. Eine Pipeline kann sich auffächern (ein Tester leitet Datensätze an zwei verschiedene Consumer) oder zusammenführen (zwei Producer speisen einen Transformer). Sie kann linear sein, was am einfachsten nachzuvollziehen ist, oder ein gerichteter azyklischer Graph von Schritten, wie in Workflow-Werkzeugen wie Apache Airflow. Streng genommen bricht ein Filter mit internem Zustand, etwa einer, der zählt oder sortiert, die inkrementelle Eigenschaft, weil er die gesamte Eingabe sehen muss, bevor er Ausgabe erzeugt; sort in einer Unix-Pipeline ist ein solcher Filter. Solche Filter sind erlaubt, aber sie werden zu den Stellen, an denen Speicher, Latenz und Neustarts Nachdenken erfordern.

Beispiele: wo Pipelines vorkommen

ETL- und Datenpipelines. ETL-Jobs (Extract, Transform, Load) bringen Daten aus operativen Systemen in ein Data Warehouse: die Bestellungen von gestern extrahieren, bereinigen und umwandeln, mit Restaurantdaten zusammenführen, in Berichtstabellen laden. Jeder Schritt ist ein Filter, und moderne Datenplattformen (Apache Beam, Spark, dbt-Modelle, Cloud-Dataflow-Dienste) sind im Wesentlichen Werkzeuge, um solche Pipelines in großem Maßstab zu bauen und zu betreiben.

Compiler. Ein klassischer Compiler ist eine Pipeline aus Phasen: Die lexikalische Analyse macht aus Zeichen Tokens, das Parsen macht aus Tokens einen Syntaxbaum, die semantische Analyse prüft Typen und Namen, die Optimierung schreibt eine Zwischendarstellung um, und die Codegenerierung erzeugt Maschinencode. Jede Phase verarbeitet eine Darstellung und erzeugt die nächste, sodass Phasen getrennt entwickelt und ersetzt werden können; das LLVM-Projekt etwa lässt viele Sprach-Frontends einen Optimierer und eine Reihe von Backends gemeinsam nutzen. Echte Compiler lockern den strengen Stil, weil Phasen eine Symboltabelle teilen.

Medienverarbeitung. Videotranskodierung und Audioverarbeitung verketten Decoder, Skalierer, Filter und Encoder. GStreamer baut Medienanwendungen ausdrücklich als Pipelines aus Elementen, die über Pads verbunden sind, und die Filtergraphen von FFmpeg funktionieren genauso.

Log- und Ereignisverarbeitung. Log-Shipper wie Logstash und Fluent Bit werden als Eingabe-, Filter- und Ausgabestufen konfiguriert: Logzeilen lesen, in Felder parsen, das Rauschen von Health-Checks verwerfen, personenbezogene Daten maskieren und alles an einen Suchindex senden. Die Observability-Daten der Essenslieferplattform fließen auf diese Weise.

Feature- und Inferenzpipelines im maschinellen Lernen. Die Aufbereitung von Trainingsdaten sowie die Vor- und Nachverarbeitung rund um ein Modell werden üblicherweise als Pipelines gebaut, worauf Kapitel 12 zurückkommt.

# Top ten URLs that returned HTTP 500 in today's access log
grep '" 500 ' access.log \
  | awk '{print $7}' \
  | sort \
  | uniq -c \
  | sort -rn \
  | head -10

import json

def read_lines(path):                 # producer
    with open(path) as f:
        for line in f:
            yield line

def parse(lines):                     # transformer
    for line in lines:
        yield json.loads(line)

def only_failed(events):              # tester
    for e in events:
        if e["status"] == "payment_failed":
            yield e

def count_by_provider(events):        # consumer
    counts = {}
    for e in events:
        counts[e["provider"]] = counts.get(e["provider"], 0) + 1
    return counts

events = only_failed(parse(read_lines("orders.jsonl")))
print(count_by_provider(events))

Batch und Streaming

Dieselbe Struktur läuft in zwei Modi. Die Stapelverarbeitung verarbeitet eine begrenzte Eingabe, etwa alle Bestellungen von gestern, läuft bis zum Ende und stoppt. Sie ist leicht nachzuvollziehen: Schlägt ein Lauf fehl, beheben Sie das Problem und starten ihn erneut, vorausgesetzt, jeder Lauf ersetzt seine Ausgabe, statt an sie anzuhängen. Nächtliche Auszahlungen an Restaurants, monatliche Rechnungen und Warehouse-Ladevorgänge sind typische Batch-Pipelines. Ihre Schwäche ist die Verzögerung: Die Ergebnisse sind so alt wie der letzte Lauf.

Stream Processing verarbeitet eine unbegrenzte Eingabe und jeden Datensatz kurz nach seinem Eintreffen: Kurierpositionen, Zahlungsereignisse, Klicks. Die Ergebnisse sind frisch, doch es tauchen neue Fragen auf. Ein Stream endet nie, daher müssen Aggregate über Zeitfenster berechnet werden (Bestellungen pro Restaurant je fünf Minuten); Datensätze können verspätet oder in falscher Reihenfolge eintreffen; und die Pipeline muss ununterbrochen laufen, sich von Abstürzen erholen, ohne Datensätze zu verlieren oder doppelt zu zählen, und Spitzen der Eingaberate verkraften. Frameworks wie Kafka Streams, Apache Flink und Apache Beam bieten Zeitfenster, Checkpoints und State Stores für diese Probleme. Das Modell von Beam behandelt Batch als Sonderfall von Streaming mit begrenzter Eingabe, sodass eine Pipeline-Definition auf beide Arten laufen kann.

Viele Organisationen betreiben beides: einen Stream für frische, näherungsweise operative Dashboards und einen Batch-Job, der über Nacht die exakten Zahlen neu berechnet. Abschnitt 7.8 behandelt die Seite der Datenarchitektur bei dieser Wahl.

Backpressure und Fehlerbehandlung

Ist ein Producer schneller als ein Consumer, muss etwas nachgeben. Die Optionen sind: puffern (was das Problem nur aufschiebt, sofern der Puffer nicht begrenzt ist), Datensätze verwerfen oder Stichproben nehmen (für manche Metriken akzeptabel, für Zahlungen nie) oder Backpressure anwenden: die schnelle Stufe auf das Tempo der langsamen bremsen. Unix-Pipes tun das automatisch, weil eine volle Pipe den Schreiber blockiert. Pull-basierte Entwürfe wie Python-Generatoren tun es konstruktionsbedingt, weil nichts erzeugt wird, bevor die nächste Stufe danach fragt. Verteilte Pipelines brauchen es als bewusstes Designelement: begrenzte Queues, Consumer, die in ihrem eigenen Tempo aus einem Log abrufen, oder Protokolle wie die Reactive-Streams-Spezifikation der JVM, in der ein Subscriber signalisiert, für wie viele Datensätze er bereit ist. Abschnitt 9.3 behandelt Queues und Backpressure ausführlich.

Fehler brauchen für jeden Filter eine Strategie, die vorab festgelegt wird. Ein fehlerhafter Datensatz (eine Poison Message) sollte die Pipeline nicht anhalten: Leiten Sie ihn mit dem Grund an eine Dead Letter Queue oder Fehlerdatei weiter, zählen Sie ihn und schlagen Sie Alarm, wenn die Rate steigt. Ein vorübergehender Fehler, etwa ein Timeout bei einem Nachschlagedienst, wird mit Back-off wiederholt. Ein Absturz der Pipeline braucht einen Wiederaufsetzpunkt: Batch-Jobs laufen von vorn oder ab einem Checkpoint neu, und Streaming-Frameworks sichern ihre Position und ihren Zustand in Checkpoints. Weil Neustarts bedeuten, dass manche Datensätze doppelt verarbeitet werden, sollten Consumer, die Ergebnisse schreiben, idempotent sein, etwa durch einen Upsert auf einen natürlichen Schlüssel statt blinden Einfügens. Bei Fehlern ist der Stil am schwächsten: Filter wissen bewusst nichts voneinander, daher kann ein in Schritt vier entdeckter Fehler nicht ohne Weiteres rückgängig machen, was Schritt zwei schon getan hat, und eine Pipeline, die ein solches koordiniertes Rollback braucht, ist vermutlich der falsche Stil.

Wann man ihn einsetzt, wann nicht, und Anti-Patterns

Setzen Sie Pipes and Filters ein, wenn das Problem von Natur aus ein Datenfluss in eine Richtung durch klar getrennte Transformationen ist: Aufnahme, Umwandlung, Anreicherung, Validierung, Aggregation, Export. Es ist die Standardwahl im Data Engineering, für die Verarbeitung von Dateien und Medien und für jeden Job, bei dem sich die Schritte häufiger ändern als der Gesamtablauf. Es kommt auch innerhalb anderer Stile vor: ein nächtlicher Reporting-Job in einem modularen Monolithen oder ein Stream-Processing-Service in einem ereignisgesteuerten System.

Vermeiden Sie es als Hauptstruktur interaktiver Anwendungen, bei denen Nutzer sofortige, dialogartige Antworten erwarten und die Arbeit kein Fluss ist; für Logik, bei der viele Schritte denselben Zustand teilen und ändern müssen; und für Prozesse, die ein koordiniertes Rollback über Schritte hinweg brauchen, was eine Transaktion oder eine Saga erfordert.

Häufige Anti-Patterns: der Gott-Filter, ein Schritt, der parst, anreichert, validiert und schreibt und damit die Vorteile des Stils verliert; verborgener gemeinsamer Zustand, Filter, die über eine gemeinsame Datenbanktabelle oder globale Variable statt über die Pipe kommunizieren und damit die Kopplung zurückbringen, die der Stil beseitigt; Formatdrift, ein Producer, der still sein Ausgabeformat ändert und jeden nachgelagerten Filter bricht, verhindert durch explizite Schemas und Schemaprüfungen (siehe Abschnitt 7.7); unbegrenzte Puffer, die einen langsamen Consumer in einen Absturz wegen Speichermangels verwandeln; und nicht idempotente Senken, die nach jeder Wiederholung doppelt zählen.

Den Stil im Code durchsetzen

Die Einschränkungen des Stils lassen sich explizit machen. Geben Sie jedem Filter dieselbe einfache Signatur (ein Iterable von Datensätzen hinein, ein Iterable heraus), damit das Zusammensetzen mechanisch wird. Definieren Sie das Datensatzformat an jeder Pipe als Schema (eine Dataclass, ein JSON Schema, eine Avro- oder Protocol-Buffers-Definition) und validieren Sie es an der Grenze. Halten Sie Filter frei von gemeinsamem veränderlichem Zustand: Ein Architekturtest kann Filtermodulen verbieten, die Datenbankschicht oder einander zu importieren, sodass Daten nur über die Pipe zwischen ihnen fließen können. Testen Sie jeden Filter mit kleinen Ein- und Ausgabe-Fixtures und die zusammengesetzte Pipeline Ende-zu-Ende mit einer repräsentativen Stichprobe, einschließlich fehlerhafter Datensätze, um zu belegen, dass die Fehlerstrategie funktioniert.

Überwachen Sie in verteilten Pipelines den Durchsatz pro Stufe, die Queue-Tiefe und den Consumer-Lag: Eine wachsende Queue vor einer Stufe ist das sichtbare Zeichen eines Engpasses oder eines versagenden Filters, und sie ist die erste Metrik, auf die man einen Alarm legen sollte.