Zurückfüllen von historischen Daten mit Pipelines

In der Datentechnik bezieht sich das Ausfüllen auf den Prozess der rückwirkenden Verarbeitung von historischen Daten über eine Datenpipeline, die für die Verarbeitung aktueller oder Streamingdaten entwickelt wurde.

Dies ist in der Regel ein separater Fluss, der Daten in Ihre vorhandenen Tabellen sendet. Die folgende Abbildung zeigt einen Backfill-Flow, der historische Daten an die Bronzetabellen in Ihrer Pipeline sendet.

Einige Szenarien, für die ein Rückfüllen erforderlich sein kann:

• Verarbeiten Sie historische Daten aus einem älteren System, um ein Ml-Modell (Machine Learning) zu trainieren oder ein historisches Trendanalysedashboard zu erstellen.
• Neu verarbeiten einer Teilmenge von Daten wegen eines Datenqualitätsproblems mit vorgelagerten Datenquellen.
• Ihre geschäftlichen Anforderungen wurden geändert, und Sie müssen Daten für einen anderen Zeitraum zurückfüllen, der nicht von der ursprünglichen Pipeline abgedeckt wurde.
• Ihre Geschäftslogik hat sich geändert, und Sie müssen sowohl historische als auch aktuelle Daten neu verarbeiten.

Ein Nachvervollständigen in Lakeflow-Pipelines wird durch einen speziellen Append-Flow unterstützt, der die Option ONCE verwendet. Für weitere Informationen zur Option siehe append_flow oder ONCE.

Überlegungen beim Zurückfüllen von historischen Daten in eine Streamingtabelle

• Normalerweise werden die Daten an die Bronze-Streaming-Tabelle angehängt. Nachgelagerte Silber- und Goldschichten nehmen die neuen Daten aus der Bronzeschicht auf.
• Stellen Sie sicher, dass Ihre Pipeline doppelte Daten ordnungsgemäß verarbeiten kann, falls dieselben Daten mehrmals angefügt werden.
• Stellen Sie sicher, dass das Verlaufsdatenschema mit dem aktuellen Datenschema kompatibel ist.
• Berücksichtigen Sie die Größe des Datenvolumes und die erforderliche Verarbeitungszeit SLA, und konfigurieren Sie entsprechend die Cluster- und Batchgrößen.

Beispiel: Hinzufügen eines Rückfüllens zu einer vorhandenen Pipeline

Angenommen, Sie haben eine Pipeline, die rohe Ereignisregistrierungsdaten aus einer Cloudspeicherquelle ab dem 01. Januar 2025 erfasst. Später stellen Sie fest, dass Sie die historischen Daten der letzten drei Jahre für nachgelagerte Berichts- und Analysezwecke wieder auffüllen möchten. Alle Daten befinden sich an einem Ort, partitioniert nach Jahr, Monat und Tag im JSON-Format.

Erste Pipeline

Hier sehen Sie den Startpipelinecode, der die rohen Ereignisregistrierungsdaten inkrementell aus dem Cloudspeicher erfasst.

Python

from pyspark import pipelines as dp

source_root_path = spark.conf.get("registration_events_source_root_path")
begin_year = spark.conf.get("begin_year")
incremental_load_path = f"{source_root_path}/*/*/*"

# create a streaming table and the default flow to ingest streaming events
@dp.table(name="registration_events_raw", comment="Raw registration events")
def ingest():
    return (
        spark
        .readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("cloudFiles.maxFilesPerTrigger", 100)
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("modifiedAfter", "2025-01-01T00:00:00.000+00:00")
        .load(incremental_load_path)
        .where(f"year(timestamp) >= {begin_year}") # safeguard to not process data before begin_year
    )

SQL

-- create a streaming table and the default flow to ingest streaming events
CREATE OR REFRESH STREAMING LIVE TABLE registration_events_raw AS
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/*/*/*",
  format => "json",
  inferColumnTypes => true,
  maxFilesPerTrigger => 100,
  schemaEvolutionMode => "addNewColumns",
  modifiedAfter => "2024-12-31T23:59:59.999+00:00"
)
WHERE year(timestamp) >= '2025'; -- safeguard to not process data before begin_year

Hier verwenden wir die modifiedAfter Option "Auto Loader", um sicherzustellen, dass wir nicht alle Daten aus dem Cloudspeicherpfad verarbeiten. Die inkrementelle Verarbeitung wird an dieser Grenze abgeschnitten.

Auffüllung von Daten aus den vergangenen 3 Jahren

Jetzt möchten Sie einen oder mehrere Datenflüsse hinzufügen, um vorherige Daten aufzufüllen. Führen Sie in diesem Beispiel die folgenden Schritte aus:

• Verwenden Sie den append once Fluss. Dadurch wird ein einmaliger Backfill durchgeführt, ohne dass die Ausführung nach diesem ersten Backfill fortgesetzt wird. Der Code verbleibt in Ihrer Pipeline und wenn die Pipeline jemals vollständig aktualisiert wird, wird das Rückfüllen erneut ausgeführt.
• Erstellen Sie drei Backfill-Flüsse, eine für jedes Jahr (in diesem Fall werden die Daten nach Jahr im Pfad aufgeteilt). Für Python parametrisieren wir die Erstellung der Flüsse, aber in SQL wiederholen wir den Code dreimal, einmal für jeden Fluss.

Wenn Sie an Ihrem eigenen Projekt arbeiten und keine serverlose Berechnung verwenden, sollten Sie die maximalen Mitarbeiter für die Pipeline aktualisieren. Indem Sie die maximalen Arbeitskräfte erhöhen, gewährleisten Sie, dass Sie über die Ressourcen verfügen, um die historischen Daten zu verarbeiten und gleichzeitig die aktuellen Streaming-Daten innerhalb des erwarteten Service Level Agreements zu verarbeiten.

Python

from pyspark import pipelines as dp

source_root_path = spark.conf.get("registration_events_source_root_path")
begin_year = spark.conf.get("begin_year")
backfill_years = spark.conf.get("backfill_years") # e.g. "2024,2023,2022"
incremental_load_path = f"{source_root_path}/*/*/*"

# meta programming to create append once flow for a given year (called later)
def setup_backfill_flow(year):
    backfill_path = f"{source_root_path}/year={year}/*/*"
    @dp.append_flow(
        target="registration_events_raw",
        once=True,
        name=f"flow_registration_events_raw_backfill_{year}",
        comment=f"Backfill {year} Raw registration events")
    def backfill():
        return (
            spark
            .read
            .format("json")
            .option("inferSchema", "true")
            .load(backfill_path)
        )

# create the streaming table
dp.create_streaming_table(name="registration_events_raw", comment="Raw registration events")

# append the original incremental, streaming flow
@dp.append_flow(
        target="registration_events_raw",
        name="flow_registration_events_raw_incremental",
        comment="Raw registration events")
def ingest():
    return (
        spark
        .readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("cloudFiles.maxFilesPerTrigger", 100)
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("modifiedAfter", "2024-12-31T23:59:59.999+00:00")
        .load(incremental_load_path)
        .where(f"year(timestamp) >= {begin_year}")
    )

# parallelize one time multi years backfill for faster processing
# split backfill_years into array
for year in backfill_years.split(","):
    setup_backfill_flow(year) # call the previously defined append_flow for each year

SQL

-- create the streaming table
CREATE OR REFRESH STREAMING TABLE registration_events_raw;

-- append the original incremental, streaming flow
CREATE FLOW
  registration_events_raw_incremental
AS INSERT INTO
  registration_events_raw BY NAME
SELECT * FROM STREAM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/*/*/*",
  format => "json",
  inferColumnTypes => true,
  maxFilesPerTrigger => 100,
  schemaEvolutionMode => "addNewColumns",
  modifiedAfter => "2024-12-31T23:59:59.999+00:00"
)
WHERE year(timestamp) >= '2025';


-- one time backfill 2024
CREATE FLOW
  registration_events_raw_backfill_2024
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2024/*/*",
  format => "json",
  inferColumnTypes => true
);

-- one time backfill 2023
CREATE FLOW
  registration_events_raw_backfill_2023
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2023/*/*",
  format => "json",
  inferColumnTypes => true
);

-- one time backfill 2022
CREATE FLOW
  registration_events_raw_backfill_2022
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2022/*/*",
  format => "json",
  inferColumnTypes => true
);

Diese Implementierung hebt mehrere wichtige Muster hervor.

Trennung von Bedenken

• Die inkrementelle Verarbeitung ist unabhängig von Backfill-Vorgängen.
• Jeder Fluss verfügt über eigene Konfigurations- und Optimierungseinstellungen.
• Es gibt einen klaren Unterschied zwischen inkrementellen und Backfill-Vorgängen.

Kontrollierte Ausführung

• Durch die Verwendung der Option ONCE wird sichergestellt, dass jeder Backfill genau einmal ausgeführt wird.
• Der Rückfüllprozess verbleibt in der Pipeline-Grafik, befindet sich danach jedoch im Leerlauf. Sie ist für die vollständige Aktualisierung automatisch einsatzbereit.
• Es gibt einen klaren Prüfpfad für Backfill-Vorgänge in der Pipelinedefinition.

Verarbeitungsoptimierung

• Sie können den großen Rückfüller zur schnelleren Verarbeitung oder zur Steuerung der Verarbeitung in mehrere kleinere Rückfüllungen aufteilen.
• Durch die Verwendung der erweiterten automatischen Skalierung wird die Clustergröße basierend auf der aktuellen Clusterlast dynamisch skaliert.

Schemaentwicklung

• Die Verwendung von schemaEvolutionMode="addNewColumns" geht mit Schemaänderungen ordnungsgemäß um.
• Sie haben konsistente Schema-Inferenz über historische und aktuelle Daten hinweg.
• Es gibt eine sichere Handhabung neuer Spalten in neueren Daten.

Beispiel: Nachvervollständigen eines SCD-Ziels während einer Migration

Ein häufiges Migrationsszenario ist eine sich langsam verändernde Dimension (SCD)-Tabelle, die bereits in einem Altsystem mit jahrelanger Geschichte existiert, dessen ursprünglicher Änderungsfeed jedoch nicht mehr verfügbar ist. Da die ursprünglichen Änderungsereignisse nicht mehr vorhanden sind, spielen Sie stattdessen die Historie der veralteten Tabelle einmalig in das neue AUTO CDC-Ziel ein und fügen anschließend einen neuen CDC-Feed für die Zukunft zu. Weitere Informationen zu AUTO CDC und SCD-Typen finden Sie unter Die AUTO-CDC-APIs: Vereinfachte Änderungsdatenerfassung mit Pipelines.

Die Struktur besteht aus einem einmaligen AUTO CDC-Flow in dieselbe Streaming-Tabelle, auf die der laufende AUTO CDC-Flow abzielt. Ein AUTO CDC-Ziel akzeptiert nur AUTO CDC-Flows, daher muss der Seed ebenfalls ein AUTO CDC-Flow sein. Ein einfacher INSERT INTO ONCE Append-Flow in dieselbe Tabelle schlägt bei der Validierung fehl:

1. Erstelle die Ziel-Streaming-Tabelle , in die dein AUTO CDC Flow schreibt.
2. Führen Sie den Seed der veralteten Historie einmalig mit einem AUTO CDC ONCE-Flow durch, der die veraltete SCD-Tabelle als Stream liest, mit einer Sequenz nach der veralteten Gültigkeitsstart-Spalte. Replayen Sie die veralteten Zeilen als Änderungsereignisse, anstatt sie selbst aufzubereiten. AUTO CDC erstellt die __START_AT- und __END_AT-Verlaufsspalten für ein SCD-Ziel vom Typ 2; schreiben Sie diese Spalten daher nicht direkt.
3. Fügen Sie den laufenden AUTO CDC-Flow hinzu, der den aktuellen Änderungs-Feed liest. AUTO CDC behebt die Reihenfolge pro Schlüssel; daher muss die Umstellung für jeden Geschäftsschlüssel einzeln gelten: Die erste Live-Änderung jedes Schlüssels muss zeitlich nach der letzten initialisierten Änderung für denselben Schlüssel erfolgen. Ein Sequenzwert, der nur später als das globale Legacy-Maximum liegt, kann für einen einzelnen Schlüssel immer noch veraltet sein, und die erste Live-Änderung dieses Schlüssels wird dann ignoriert oder falsch geordnet.

Der folgende Code erstellt eine Streaming-Tabelle, die die obigen Schritte verwendet:

CREATE OR REFRESH STREAMING TABLE customers_history;

-- One-time seed: replay the legacy history as change events
CREATE FLOW customers_history_seed
AS AUTO CDC ONCE INTO customers_history
FROM stream(legacy.customers_scd2)
KEYS (customer_id)
SEQUENCE BY valid_from
STORED AS SCD TYPE 2;

-- Ongoing live CDC into the same target
CREATE FLOW customers_history_cdc
AS AUTO CDC INTO customers_history
FROM stream(customers_cdc_bronze)
KEYS (customer_id)
SEQUENCE BY change_timestamp
STORED AS SCD TYPE 2;

Beide Datenflüsse müssen bei ihren Schlüsseln, ihrem SCD-Typ und dem Datentyp ihrer Sequenzspalte übereinstimmen. Im vorangegangenen Beispiel werden beide Flows anhand eines Zeitstempels sequenziert, wobei ein einziger Cutover-Zeitpunkt verwendet wird, um die vorab eingegebenen historischen Daten vom Live-Feed zu trennen. Wenn die veraltete Tabelle anhand eines Werts eines anderen Typs als der Live-Feed sequenziert wird, wandeln Sie einen der beiden Typen so um, dass die Typen übereinstimmen.

Die gleiche Form funktioniert für ein SCD-Typ-1-Ziel: Ändern Sie in beiden Flüssen STORED AS SCD TYPE 2 in STORED AS SCD TYPE 1, und das Ziel behält pro Schlüssel nur die aktuelle Zeile. Bevor Sie sich auf eine der beiden Varianten verlassen, prüfen Sie anhand einer Stichprobe von Schlüsseln, dass die erste Änderung im Live-Betrieb für einen vorinitialisierten Schlüssel genau eine neue Version erzeugt und die vorherige korrekt abschließt. Eine Sequenzierungslücke pro Schlüssel tritt in der Regel in diesem Schritt auf.

Weitere Ressourcen

• Inkrementelles Laden und Verarbeiten von Daten mit Lakeflow-Pipelineflüssen
• Verwenden Sie Flows in Lakeflow-Pipelines
• append_flow
• CREATE FLOW (Pipelines)