Verarbeitungsgarantien in Lakeflow-Pipelines

Wiederholungsversuche und erneute Ausführungen sind in jeder realen Pipeline unvermeidlich. Daher erläutert diese Seite die Verarbeitungsgarantien, die Lakeflow-Pipelines bieten, und wie Sie sicherstellen können, dass die von Ihnen geschriebenen Elemente sicher erneut ausgeführt werden können.

Overview

Zwei verwandte Eigenschaften bestimmen, ob das Wiederverlegen einer Pipeline sicher ist:

  • Idempotenz bedeutet, dass eine Pipeline dasselbe Ergebnis liefert, egal wie oft man sie über denselben Eingang ausführt. Eine erneute Ausführung nach einem Fehler, das doppelte Nachholen eines Datumsbereichs oder das manuelle erneute Auslösen eines Jobs erstellt niemals doppelte Zeilen oder führt zu einer fehlerhaften Status-Beschädigung.
  • Die Verarbeitungsgarantie beschreibt, wie oft jeder Datensatz das Ergebnis beeinflusst. Mindestens einmal wird garantiert, dass jeder Datensatz verarbeitet wird, aber ein Fehler und ein erneuter Versuch können einige Datensätze mehr als einmal verarbeiten, was das Risiko von Doppelgängern bedeutet. Eine exakt einmalige Verarbeitung garantiert, dass jeder Datensatz das Ergebnis beeinflusst, als wäre er genau einmal verarbeitet worden, selbst über Wiederholungen hinweg, ohne Duplikate und ohne Lücken.

Lakeflow-Pipelines sind standardmäßig idempotent für die von ihnen verwalteten Teile und bieten Ihnen eine „Exactly-Once“-Verarbeitung innerhalb ihrer eigenen verwalteten Tabellen. Wichtig zu verstehen ist, wo diese Garantien aufhören, automatisch zu sein, damit Sie die richtigen Schutzmaßnahmen an den Rändern Ihrer Pipeline einbauen können.

So funktioniert es

Lakeflow-Pipelines bieten „Exactly-Once“-Verarbeitung und Idempotenz für die von ihnen verwalteten Flows und stellen Ihnen Tools zur Verfügung, um auch die von Ihnen geschriebene Logik idempotent zu halten.

Genau-einmal-Verarbeitung für verwaltete Tabellen

In verwalteten Tabellen ist standardmäßig Exactly-once-Verarbeitung verfügbar. Streaming-Tabellen nutzen Structured-Streaming-Checkpoints in Kombination mit den transaktionalen Schreibvorgängen von Delta Lake: Jeder Mikro-Batch committet seine Quell-Offsets und seine Ausgabe gemeinsam, sodass ein nach einem Fehler wiederholter Batch entweder vollständig erfolgreich ist oder vollständig einen Roll-Back durchläuft und erneut ausgeführt wird – er wird niemals teilweise zweimal angewendet. Dies gilt für die Dateiingestion mit Auto Loader, für Lesevorgänge aus Kafka, Kinesis und Azure Event Hubs sowie für AUTO CDC Upserts – ganz ohne, dass du Code schreiben musst.

Wenn eine mindestens einmalige Quelle denselben Datensatz mehrfach sendet, verarbeitet die Pipeline sie als eindeutige Datensätze und schreibt alle in deine Tabelle. Das Entfernen dieser Duplikate liegt in Ihrer Verantwortung. Siehe Quellen mindestens einmal deduplizieren.

Die Idempotenz beim Lesen ergibt sich aus denselben Checkpoints. Auto Loader und Checkpoints von Streaming-Tabellen gewährleisten, dass jede Quelldatei oder jeder Offset zur Zustandsverfolgung einmal verarbeitet wird; somit wird die erneute Verarbeitung einer Pipeline-Aktualisierung nach einem Fehler ab dem Checkpoint fortgesetzt, anstatt Daten erneut zu verarbeiten oder zu überspringen. Dies erreichen Sie durch die Verwendung von Streaming-Tabellen über spark.readStream anstelle von manuell erstellten Batch-Schleifen. Siehe Streamingtabellen.

Verwenden Sie AUTO CDC anstelle eines manuell geschriebenen MERGE

AUTO CDC INTO ist inhärent idempotent in Bezug auf seine keys und sequence_by. Das zweimalige Anwenden desselben Änderungsdatensatzes oder das Anwenden von Datensätzen in der falschen Reihenfolge erzeugt denselben Endzustand, weil die Pipeline die Sequenzspalte verwendet, um zu entscheiden, ob eine eingehende Zeile tatsächlich neuer ist als die gespeicherte:

CREATE FLOW customers_cdc_flow AS AUTO CDC INTO customers_silver
FROM stream(customers_cdc_bronze)
KEYS (customer_id)
SEQUENCE BY sequence_num
STORED AS SCD TYPE 1;

Wenn du deine eigene Upsert-Logik außerhalb von AUTO CDC schreibst (selten, aber manchmal für komplexe Merge-Bedingungen erforderlich), richte sie an einem stabilen Geschäftsschlüssel aus und stelle sicher, dass sie problemlos zweimal angewendet werden kann, zum Beispiel ein MERGE ... WHEN MATCHED, das über order_id geschlüsselt ist, anstelle eines blinden INSERT. Weitere Informationen finden Sie unter Die AUTO CDC-APIs: Vereinfachen Sie Change Data Capture mit Pipelines.

Sorgen Sie dafür, dass Ihre eigenen Transformationen idempotent bleiben.

Damit die Logik beim erneuten Ausführen von Schreibvorgängen idempotent bleibt, befolgen Sie diese beiden Richtlinien:

  • Vermeiden Sie nichtdeterministische Transformationen in materialisierten Ansichten. Da eine materialisierte Ansicht vollständig oder inkrementell neu berechnen kann, sollte man Funktionen vermeiden, deren Ausgabe davon abhängt , wann sie ausgeführt werden, statt von der Eingabe . Verwenden Sie beispielsweise nicht current_timestamp(), um einen geschäftlichen Wert zu berechnen, der nach dem Schreiben unverändert bleiben soll; verwenden Sie den Zeitstempel aus dem Quellereignis oder übergeben Sie ihn als Parameter, sodass eine Neuberechnung eine identische Ausgabe erzeugt.
  • Entwickle vollständige Aktualisierungen, um auf Nummer sicher zu gehen. Eine vollständige Aktualisierung löscht eine Tabelle und berechnet sie von Grund auf neu, was nur dann sicher ist, wenn jede vorgelagerte Quelle noch die gesamte Historie erzeugen kann. Wenn eine vorgelagerte Quelle nur ein rollierendes Fenster von Änderungen zur Verfügung stellt, kann bei einer vollständigen Aktualisierung einer nachgelagerten AUTO CDC-Tabelle die Historie unbemerkt verloren gehen; berücksichtigen Sie dies daher bei der Gestaltung der Aufbewahrung für Quellen und Themen.

„Exactly-once“ an den Rändern gewährleisten

„Exactly-once“ hört dort auf, automatisch zu funktionieren, wo die Pipeline keinen direkten Einfluss mehr hat, wie z. B. bei Schreibvorgängen in externe Systeme. Wenn Sie Daten an ein externes System weiterleiten, gestalten Sie den Schreibvorgang selbst idempotent, beispielsweise durch „Upsert“ nach Schlüssel auf der Empfängerseite, da ein wiederholter Mikro-Batch andernfalls denselben Batch zweimal schreiben könnte. Der folgende Sink schreibt jede Partition des Batches aus den Executor-Servern und verwendet einen Idempotenzschlüssel, damit ein wiederholter Batch nicht doppelt geschrieben wird:

from pyspark import pipelines as dp

@dp.foreach_batch_sink(name="orders_to_external_api")
def write_orders_to_api(batch_df, batch_id):
    def write_partition(rows):
        # Open one client per partition.
        for row in rows:
            # Use an idempotency key (order_id) so a retried batch doesn't double-write.
            upsert_to_external_system(key=row.order_id, payload=row.asDict())

    batch_df.select("order_id", "amount").foreachPartition(write_partition)

Weitere Informationen zum Schreiben in externe Systeme finden Sie unter Sinks in Lakeflow-Pipelines.

Deduplizieren von „At-Least-Once“-Quellen

Wenn eine Quelle einen Datensatz mehr als einmal liefern kann, deduplizieren Sie downstream. Kombiniere ein Wasserzeichen mit dropDuplicatesWithinWatermark, das wasserzeichen-bewusst ist und keinen unbeschränkten Zustand benötigt, um Duplikate zu erkennen. Führen Sie die Deduplizierung anhand der Spalten durch, die ein Ereignis eindeutig identifizieren. Die Identität kann mehrere Spalten umfassen, wenn keine einzelne Spalte für sich allein eindeutig ist. Im folgenden Beispiel ist eine Klicksequenznummer nur innerhalb ihrer Sitzung eindeutig, sodass die beiden Spalten zusammen das Ereignis identifizieren:

from pyspark import pipelines as dp

@dp.table(name="clicks_deduped")
def clicks_deduped():
    return (
        spark.readStream.table("clicks_bronze")
        .withWatermark("click_ts", "5 minutes")
        .dropDuplicatesWithinWatermark(["session_id", "click_seq_num"])
    )

Wählen Sie diese Spalten aus dem Eindeutigkeitsvertrag der Quelle, nicht nach dem, was in den Beispieldaten unterschiedlich aussieht. Spalten, die sich legitimerweise wiederholen können, verwerfen echte Ereignisse, wenn Sie sie als Identitätsmerkmal behandeln. Ein gängiges Beispiel ist ein Benutzer, der zweimal auf dieselbe Anzeige klickt: Durch die Deduplizierung anhand des Benutzers und der Anzeige wird der zweite Klick stillschweigend verworfen.

Die schlüsselbasierte „Upsert“-Semantik von AUTO CDC reduziert Duplikate zudem auf natürliche Weise, sodass das Leiten von „At-Least-Once“-Daten durch einen AUTO CDC-Flow, der auf einen stabilen Geschäftsschlüssel indexiert ist, eine weitere Möglichkeit darstellt, einen „Exactly-Once“-Zustand zu erreichen.

Einschränkungen

Die „Exactly-Once“-Verarbeitung gilt für verwaltete Delta-zu-Delta-Flows. Behandeln Sie die folgenden Kanten als „at-least-once“ und fügen Sie dort explizite Deduplizierungs- oder idempotente Schreiblogik hinzu:

  • foreach_batch_sink und angepasste externe Schreibvorgänge. Spark garantiert, dass ein Batch mindestens einmal versucht wird, aber ein Batch, der nach einem teilweisen Schreiben erneut versucht wird, kann einige Zeilen im externen System zweimal sichtbar lassen. Machen Sie den externen Schreibvorgang idempotent, beispielsweise durch „Upsert“ anhand eines natürlichen Schlüssels oder durch das Schreiben einer Batch-ID, anhand derer der Empfänger Duplikate entfernen kann.
  • Kafka als Sink. Kafka-Themen unterstützen keine transaktionalen „Exactly-Once“-Schreibvorgänge in der Weise, wie Delta dies tut; daher kann ein wiederholter Mikro-Batch-Schreibvorgang an Kafka zu doppelten Nachrichten führen. Wenn nachgelagerte Verbraucher empfindlich auf Duplikate reagieren, führen Sie die Deduplizierung auf der Verbraucherseite durch, beispielsweise anhand der Ereignis-ID.
  • Benutzerdefinierte Python-Datenquellen, die als Quellen verwendet werden. Ob Lesevorgänge „Exactly-Once“ erfolgen, hängt davon ab, ob Ihre Quellimplementierung Offsets korrekt liefert und die Verarbeitung ab diesen Offsets fortsetzt. Falls sie Offsets nicht nachverfolgt, behandeln Sie sie als „at-least-once“ und führen Sie die Deduplizierung nachgelagert mit dropDuplicates anhand einer Ereignis-ID durch oder nutzen Sie die schlüsselbasierte „Upsert“-Semantik von AUTO CDC.

Als Faustregel gilt: Wenn Ihre gesamte Pipeline von Delta zu Delta verläuft (Streaming-Tabellen und materialisierte Ansichten, die Delta-Tabellen über verwaltete Flows lesen und schreiben), verfügen Sie bereits über „exactly-once“. Sobald Sie ein foreach_batch_sink, eine Nicht-Delta-Senke oder eine nicht verifizierte benutzerdefinierte Quelle hinzufügen, behandeln Sie diese spezifische Kante als „at-least-once“ und fügen Sie dort idempotente Schreib- oder Deduplizierungslogik ein.

Weitere Ressourcen

  • Last updated on