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.

Der verwendete Backfill-Flow hängt von der Zieltabelle und den Quelldaten ab. Für ein AUTO CDC langsam veränderndes Dimensionsziel (SCD) Typ-1 mit einem autoritativen Snapshot verwenden Sie einen einmaligen AUTO CDC FROM SNAPSHOT Fluss. Für eine SCD-Migration, die historische Änderungen erneut abspielt, verwenden Sie einen einmaligen AUTO CDC-Fluss.

Für Append-only-Backfills: Verwenden Sie einen spezialisierten Append-Flow mit der Option ONCE, um eine Append-only-Streaming-Tabelle backzufüllen. 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. Die nachgeschalteten 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.
• Betrachten Sie die Datenvolumengröße und das erforderliche Service-Level-Agreement (SLA) und konfigurieren Sie entsprechend die Cluster und die Batch-Größen.

Beispiel: Fügen Sie einer bestehenden Pipeline einen Backfill hinzu

Angenommen, Sie haben eine Pipeline, die rohe Ereignisregistrierungsdaten aus einer Cloudspeicherquelle ab dem 01. Januar 2025 erfasst. Später merkt man, dass man die vorherigen drei Jahre historischer Daten für nachgelagerte Berichts- und Analyseanwendungen rückwirkend befüllen möchte. 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.

Tipp

Andere Datenquellen wie Kafka, Kinesis und Azure Event Hubs verfügen über entsprechende Leseoptionen, um dasselbe Verhalten zu erzielen.

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.

Tipp

Wenn Sie serverlose Berechnung mit erweiterter automatischer Skalierung (Standardeinstellung) verwenden, erhöht sich ihr Cluster automatisch in der Größe, wenn die Auslastung zunimmt.

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.

Füge einer AUTO CDC SCD Typ 1 Tabelle eine Rückfüllung hinzu

Verwenden Sie einen einmaligen AUTO CDC FROM SNAPSHOT Flow, um einen autoritativen Schnappschuss zu einem SCD Typ-1-Ziel hinzuzufügen, das auch einen laufenden Change Data Capture (CDC)-Feed erhält. Die Snapshot-Version und die CDC-Sequenzspalte bilden eine Ordnungsdomäne. Ein neueres CDC-Ereignis hat Vorrang vor einem älteren Snapshot, während ein neueres Snapshot Vorrang vor einem älteren CDC-Ereignis hat.

Requirements

Bevor Sie das Backfill hinzufügen, stellen Sie sicher, dass die Flows die folgenden Anforderungen erfüllen:

• Das Ziel verwendet SCD Type 1.
• Das Target hat genau einen AUTO CDC FROM SNAPSHOT Flow und einen oder mehrere eindeutig benannte AUTO CDC Flows.
• Alle Flows verwenden die gleiche Anzahl von Schlüsseln in derselben Reihenfolge. Snapshot-Flow-Schlüsselnamen werden groß- und kleinschreibungsunsensitiv mit AUTO CDC Schlüsselnamen verglichen. Mehrere Flows AUTO CDC müssen identische Schlüsselnamen und Groß- und Kleinschreibung verwenden.
• Die Snapshot-Version und jede CDC-Sequenzspalte haben exakt denselben Datentyp.
• Der Flow AUTO CDC FROM SNAPSHOT definiert keine Erwartungen.
• Die Flows AUTO CDC verwenden IGNORE NULL UPDATES nicht. In Python setzen Sie ignore_null_updates_column_list, ignore_null_updates_except_column_list oder ignore_null_updates nicht.
• Die Pipeline verwendet den Trigger-Modus. Dieses Muster unterstützt keine kontinuierlichen Pipelines.

Beide Flow-Typen können die SQL- oder Python-Pipeline-Schnittstelle verwenden. Du kannst SQL- und Python-Flows im selben Ziel mischen.

Der Schnappschuss muss den vollständigen Zustand der Quelle in ihrer Version darstellen. Wenn ein Zielschlüssel im Snapshot fehlt, AUTO CDC FROM SNAPSHOT behandelt das Fehlen bei der Snapshot-Version als Löschung. Ein CDC-Ereignis mit einer neueren Version erhält oder stellt den Schlüssel wieder her.

Fügen Sie den Backfill hinzu

Um einen einmaligen Snapshot-Backfill hinzuzufügen und die CDC-Ereignisse weiter zu verarbeiten, verwenden Sie die folgenden Schritte:

1. Behalten Sie die bestehende Zieltabelle und ihre laufenden AUTO CDC Flüsse in der Pipeline-Definition.
2. Definieren Sie den autoritativen Schnappschuss und seine Version. Für einen Python-Callback muss der erste Aufruf einen Snapshot und eine Version zurückgeben. Nur None zurückkehren, nachdem mindestens ein Snapshot verarbeitet wurde.
3. Füge einen AUTO CDC FROM SNAPSHOT Flow mit once=True in Python oder ONCE in SQL hinzu. Für ein SQL-Backfill in ein bestehendes Ziel fügen Sie eine WITH VERSION Abfrage hinzu. Ein SQL-Snapshot-Flow ohne WITH VERSION unterstützt nur eine anfängliche Ladung in ein leeres Ziel.
4. Führen Sie ein ausgelöstes Pipeline-Update durch, um das Backfill und laufende CDC-Ereignisse zu verarbeiten.

Das folgende Beispiel beginnt mit einer bestehenden Python-Pipeline, die Änderungen von customers_cdc schrittweise verarbeitet. Angenommen, Sie haben diese Pipeline bereits ausgeführt und das customers Ziel befüllt:

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def customers_cdc():
    return (
        spark.readStream.table("main.bronze.customers_cdc")
        .withColumn("change_timestamp", col("change_timestamp").cast("timestamp"))
    )


dp.create_streaming_table("customers")

dp.create_auto_cdc_flow(
    name="customers_incremental_cdc",
    target="customers",
    source="customers_cdc",
    keys=["customer_id"],
    sequence_by=col("change_timestamp"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "change_timestamp"],
    stored_as_scd_type=1,
)

Um dieses bestehende Ziel rückwirkend mit dem Stand von customers_snapshot zum 1. Januar 2025 zu befüllen, fügen Sie den folgenden Code zur gleichen Pipeline-Definition hinzu. Behalten Sie die bestehende Zieltabelle und den AUTO CDCAblauf bei:

from datetime import datetime, timezone
from typing import Optional, Tuple

from pyspark.sql import DataFrame

backfill_version = datetime(2025, 1, 1, tzinfo=timezone.utc)


def backfill_snapshot_and_version(
    latest_snapshot_version: Optional[datetime],
) -> Optional[Tuple[DataFrame, datetime]]:
    if latest_snapshot_version is None:
        return (spark.read.table("main.legacy.customers_snapshot"), backfill_version)
    return None


dp.create_auto_cdc_from_snapshot_flow(
    target="customers",
    source=backfill_snapshot_and_version,
    keys=["customer_id"],
    stored_as_scd_type=1,
    once=True,
)

Der Callback muss bei seiner ersten Aufrufung einen Snapshot und eine Version zurückgeben. Wenn None zurückgegeben wird, bevor ein Snapshot verarbeitet wurde, scheitert das Pipeline-Update. Nachdem der Schnappschuss verarbeitet wurde, signalisiert die Rückgabe von None, dass keine weiteren Snapshots verfügbar sind.

Die Snapshot-Version ist ein Python-datetime, die dem Spark-SQL-Typ TIMESTAMP entspricht. Der vorhandene AUTO CDC Fluss wird in TIMESTAMP umgewandelt, sodass die beiden Sequenzierungstypen exakt übereinstimmen. Das Beispiel verwendet Python für beide Flows, aber man kann jeden der beiden Flows in SQL definieren und SQL- sowie Python-Flows im selben Ziel mischen. Für SQL-Syntax, einschließlich der erforderlichen WITH VERSION Abfrage für ein nicht-leeres Ziel, siehe CREATE FLOW (pipelines).

Nachdem der Snapshot-Fluss erfolgreich committiert ist, überspringen nachfolgende inkrementelle Updates ihn, während der AUTO CDC-Flow weiterhin neue Ereignisse verarbeitet.

Important

Eine vollständige Aktualisierung des Ziels führt den einmaligen Snapshot-Fluss erneut aus. Halte den Snapshot verfügbar und stelle sicher, dass er noch den vorgesehenen Zustand repräsentiert, bevor du eine vollständige Aktualisierung durchführst.

Dieses einheitliche Backfill-Muster unterstützt keine SCD Typ 2 oder bitemporale Ziele.

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 verschwunden sind, wird stattdessen die eigene Historie der Legacy-Tabelle einmal in das neue AUTO CDC Ziel abgespielt, und dann wird ein neuer CDC-Feed angehängt. 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. Initialisiere die Legacy-Historie einmalig mit einem AUTO CDC ONCE Flow, der die Legacy-SCD-Tabelle als Stream liest, sequenziert nach der Legacy-Gültigkeitsstartspalte. 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
• Die AUTO CDC-APIs: Vereinfachen der Änderungsdatenerfassung mit Pipelines
• append_flow
• create_auto_cdc_from_snapshot_flow
• create_auto_cdc_flow
• CREATE FLOW (Pipelines)