Återskapa historiska data med pipelines

Inom datateknik refererar återfyllnad till processen att retroaktivt bearbeta historiska data via en datapipeline som har utformats för bearbetning av aktuella eller strömmande data.

Det här är vanligtvis ett separat flöde som skickar data till dina befintliga tabeller. Följande bild visar ett återfyllnadsflöde som skickar historiska data till bronstabellerna i pipelinen.

Vissa scenarier som kan kräva en återfyllnad:

• Bearbeta historiska data från ett äldre system för att träna en maskininlärningsmodell (ML) eller skapa en instrumentpanel för historisk trendanalys.
• Bearbeta en delmängd av data på grund av ett datakvalitetsproblem med överordnade datakällor.
• Dina affärskrav har ändrats och du måste fylla på data under en annan tidsperiod som inte omfattades av den inledande pipelinen.
• Din affärslogik har ändrats och du måste bearbeta om både historiska och aktuella data.

Backfill-flödet du använder beror på måltabellen och källdatan. För en långsamt föränderlig dimension AUTO CDC av typ 1 med en auktoritativ ögonblicksbild använder du ett engångsflöde AUTO CDC FROM SNAPSHOT. För en SCD-migration som återskapar historiska förändringar, använd ett engångsflöde AUTO CDC .

För append-only backfills: Använd ett specialiserat appendflöde med alternativet ONCE för att backfylla en append-only streamingtabell. Mer information om alternativet finns i append_flow eller ONCE).

Överväganden vid återfyllning av historiska data i en strömningstabell

• Vanligtvis läggs data till i bronstabellen för strömmande data. Nedströmsliggande silver- och guldlager plockar upp den nya datan från bronslagret.
• Se till att din pipeline kan hantera duplicerade data korrekt om samma data läggs till flera gånger.
• Kontrollera att det historiska dataschemat är kompatibelt med det aktuella dataschemat.
• Ta hänsyn till datavolymens storlek och det nödvändiga servicenivåavtalet (SLA) för bearbetningen, och konfigurera därefter klustret och batchstorlekarna.

Exempel: Lägg till en backfill till en befintlig pipeline

I det här exemplet säger du att du har en pipeline som matar in råa händelseregistreringsdata från en molnlagringskälla från och med jan 01, 2025. Senare inser du att du vill fylla på tidigare tre års historiska data i efterhand för efterföljande rapportering och analys. Alla data finns på en plats, partitionerade efter år, månad och dag, i JSON-format.

Inledande pipeline

Här är den start-pipelinekod som inkrementellt matar in rådata för händelseregistrering från molnlagringen.

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

Här använder vi modifiedAfter Auto Loader-alternativet för att säkerställa att vi inte bearbetar alla data från molnlagringens sökväg. Den inkrementella bearbetningen begränsas vid den gränsen.

Återfyllnadsdata från tidigare 3 år

Nu vill du lägga till ett eller flera flöden för att fylla på tidigare data. I det här exemplet utför du följande steg:

• Använd append once flödet. Detta utför en engångsåterfyllnad utan ytterligare körning efter den inledande återfyllningen. Koden finns kvar i pipelinen och om pipelinen någonsin återställs helt körs återfyllningen igen.
• Skapa tre återfyllnadsflöden, en för varje år (i det här fallet delas data efter år i sökvägen). För Python parameteriserar vi skapandet av flödena, men i SQL upprepar vi koden tre gånger, en gång för varje flöde.

Om du arbetar med ditt eget projekt och inte använder serverlös beräkning kanske du vill uppdatera maximalt antal arbetare för pipelinen. Om du ökar maxantalet ser du till att du har resurser för att bearbeta historiska data samtidigt som du fortsätter att bearbeta aktuella strömmande data i det förväntade serviceavtalet.

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
);

Den här implementeringen belyser flera viktiga mönster.

Separering av ansvar

• Inkrementell bearbetning är oberoende av datapåfyllningsoperationer.
• Varje flöde har sina egna konfigurations- och optimeringsinställningar.
• Det finns en tydlig skillnad mellan inkrementella åtgärder och återfyllnadsåtgärder.

Kontrollerad exekvering

• Om du använder alternativet ONCE ser du till att varje återfyllnad körs exakt en gång.
• Återfyllnadsflödet finns kvar i pipelinediagrammet, men blir inaktivt när det är klart. Den är redo att användas vid fullständig uppdatering, automatiskt.
• Det finns ett tydligt revisionsspår med återfyllnadsoperationer i pipeline-definitionen.

Bearbetningsoptimering

• Du kan dela upp den stora bakfyllnaden i flera mindre bakfyllningar för snabbare bearbetning eller bättre kontroll över bearbetningen.
• Med förbättrad autoskalning skalas klusterstorleken dynamiskt baserat på den aktuella klusterbelastningen.

Schemautveckling

• Använda schemaEvolutionMode="addNewColumns" hanterar schemaändringar på ett korrekt sätt.
• Du har konsekventa schemainferenser för historiska och aktuella data.
• Det finns en säker hantering av nya kolumner i nyare data.

Lägg till en backfill till en AUTO CDC SCD Typ 1-tabell

Använd ett engångsflöde AUTO CDC FROM SNAPSHOT för att lägga till en auktoritativ ögonblicksbild till ett SCD Type 1-mål som också får ett löpande CDC-flöde. Snapshot-versionen och CDC-sekvenskolumnen bildar en ordningsdomän. En nyare CDC-händelse går före en äldre snapshot, medan en nyare snapshot går före en äldre CDC-händelse.

Krav

Innan du lägger till backfill, se till att flödena uppfyller följande krav:

• Målet använder SCD typ 1.
• Målet har exakt ett AUTO CDC FROM SNAPSHOT flöde och ett eller flera unikt namngivna AUTO CDC flöden.
• Alla flöden använder samma antal nycklar i samma ordning. Snapshot-flödesnyckelnamn jämförs utan hänsyn till versaler/gemener med AUTO CDC nyckelnamn. Flera AUTO CDC flöden måste använda identiska nyckelnamn och versaler/gemener.
• Snapshot-versionen och varje CDC-sekvenseringskolumn har exakt samma datatyp.
• Flödet AUTO CDC FROM SNAPSHOT definierar inte förväntningar.
• Flödena AUTO CDC använder inte IGNORE NULL UPDATES. I Python ska du inte ange ignore_null_updates, ignore_null_updates_column_list eller ignore_null_updates_except_column_list.
• Pipelinen använder utlöst läge. Detta mönster stöder inte kontinuerliga pipelines.

Båda flödestyperna kan använda SQL-gränssnittet för pipelines eller Python-gränssnittet för pipelines. Du kan blanda SQL- och Python-flöden i samma mål.

Ögonblicksbilden måste representera källans fullständiga tillstånd i dess version. Om en målnyckel saknas i snapshoten, behandlar AUTO CDC FROM SNAPSHOT frånvaron som en borttagning i snapshot-versionen. En CDC-händelse med en nyare version bevarar eller återställer nyckeln.

Lägg till backfill

För att lägga till en engångs-snapshot backfill och fortsätta bearbeta CDC-händelser, använd följande steg:

1. Behåll den befintliga måltabellen och dess löpande AUTO CDC flöden i pipeline-definitionen.
2. Definiera den auktoritativa snapshoten och dess version. För en Python-callback måste den första anropningen returnera en snapshot och version. Returnera None endast efter att minst en snapshot har behandlats.
3. Lägg till ett AUTO CDC FROM SNAPSHOT flöde med once=True i Python eller ONCE i SQL. För en SQL-backfill i ett befintligt mål, inkludera en WITH VERSION fråga. Ett SQL-snapshotflöde utan WITH VERSION stöder endast en initial inläsning till ett tomt mål.
4. Kör en triggad uppdatering av pipeline för att hantera backfill och pågående CDC-händelser.

Följande exempel börjar med en befintlig Python-pipeline som inkrementellvis bearbetar ändringar från customers_cdc. Antag att du redan har kört denna pipeline och fyllt målet customers:

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,
)

För att bakåtfylla det befintliga målet med tillståndet för customers_snapshot per 1 januari 2025, lägg till följande kod i samma pipeline-definition. Behåll den befintliga måltabellen och AUTO CDC flödet:

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,
)

Callbacken måste returnera en snapshot och version vid sitt första anrop. Om den returnerar None innan någon snapshot har bearbetats, misslyckas pipelineuppdateringen. Efter att snapshoten har bearbetats signalerar att returnera None att inga ytterligare snapshots finns tillgängliga.

Snapshot-versionen är en Python datetime, vilket motsvarar Spark SQL-typen TIMESTAMP. Det befintliga AUTO CDC-flödet typomvandlar change_timestamp till TIMESTAMP så att de två sekvenseringstyperna matchar exakt. Exemplet använder Python för båda flödena, men du kan definiera endera flödet i SQL och blanda SQL- och Python-flöden i samma mål. För SQL-syntax, inklusive den nödvändiga WITH VERSION frågan för en icke-tom måltabell, se CREATE FLOW (pipelines).

Efter att snapshot-flödet har kommittats framgångsrikt, hoppar efterföljande inkrementella uppdateringar över det medan AUTO CDC flödet fortsätter att bearbeta nya händelser.

Detta enhetliga backfill-mönster stöder inte SCD Typ 2 eller bitemporala mål.

Exempel: Fyll på ett SCD-mål under en migration

Ett vanligt migrationsscenario är en långsamt förändrande dimension (SCD)-tabell som redan finns i ett äldre system med flera års ackumulerad historik, men vars ursprungliga ändringsflöde inte längre är tillgängligt. Eftersom de ursprungliga ändringshändelserna är borta, spelar du istället upp den äldre tabellens egen historik i det nya AUTO CDC målet en gång, och ansluter sedan ett nytt CDC-flöde framöver. För mer om AUTO CDC och SCD-typer, se AUTO CDC API:er: Förenkla förändringsdatainsamling med pipelines.

Mönstret är ett engångs-AUTO CDC-flöde in i samma strömmande tabell som det pågående AUTO CDC-flödet är riktat mot. Ett AUTO CDC mål accepterar endast AUTO CDC flöden, så fröet måste också vara ett AUTO CDC flöde. Ett vanligt INSERT INTO ONCE appendflöde i samma tabell misslyckas med valideringen:

1. Skapa målströmningstabellen som ditt AUTO CDC flöde skriver till.
2. Initiera den äldre historiken en gång med ett AUTO CDC ONCE flöde som läser den äldre SCD-tabellen som en ström, sekvenserad av den äldre giltighetsstartkolumnen. Spela om de gamla raderna som förändringshändelser istället för att forma dem själv. AUTO CDC skapar historikkolumnerna __START_AT och __END_AT för ett SCD typ 2-mål, så skriv inte dessa kolumner direkt.
3. Anslut det pågående AUTO CDC flödet som läser den senaste ändringsfeeden. AUTO CDC hanterar ordningen per nyckel, så övergången måste upprätthållas för varje affärsnyckel individuellt: varje nyckels första ändring i drift måste komma i sekvens efter den sista inlästa ändringen för samma nyckel. Ett sekvensvärde som bara är senare än det globala äldre maxvärdet kan fortfarande vara föråldrat för en enskild nyckel, och den nyckelns första levande ändring ignoreras eller ordnas fel.

Följande kod skapar en strömningstabell som använder ovanstående steg:

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;

Båda flödena måste vara överens om sina nycklar, sin SCD-typ och datatypen för sin sekvenseringskolumn. I föregående exempel sekvenseras båda flödena med en tidsstämpel, som använder en enda cutover-tid för att separera den seedade historiken från liveflödet. Om sekvenserna i den äldre tabellen sorteras efter ett värde av en annan typ än liveflödet, typomvandla ett av dem så att typerna matchar.

Samma form fungerar för ett SCD Type 1-mål: byt STORED AS SCD TYPE 2 till STORED AS SCD TYPE 1 på båda flödena, och målet behåller endast den aktuella raden per tangent. Innan du förlitar dig på någon av formerna, verifiera på ett prov av nycklar att den första live-ändringen för en seedad nyckel ger exakt en ny version och korrekt stänger den föregående. Ett sekvensgap per nyckel uppstår vanligtvis vid det steget.

Ytterligare resurser

• Läsa in och bearbeta data stegvis med Lakeflow-pipelineflöden
• Använd flöden i Lakeflow-pipelines
• API:er för AUTOMATISK CDC: Förenkla insamling av ändringsdata med pipelines
• append_flow
• create_auto_cdc_from_snapshot_flow
• create_auto_cdc_flow
• CREATE FLOW (pipelines)