Å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.

Bakåtfyllning i Lakeflow-pipelines stöds genom ett specialiserat append-flöde som använder alternativet ONCE. 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öms silver- och guldlager hämtar nya data från bronsskiktet.
• 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.
• Överväg datavolymstorleken och det nödvändiga serviceavtalet för bearbetningstid och konfigurera därefter klustret och batchstorlekarna.

Exempel: Lägga till en återfyllnad i 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. Du inser senare att du vill fylla på historiska data från de föregående tre åren för nedströmsrapportering och analysanvändningsfall. 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.

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 bifogar 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 befintliga historiken en gång med ett AUTO CDC ONCE-flöde som läser den befintliga SCD-tabellen som en ström, sorterat efter den befintliga 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
• append_flow
• CREATE FLOW (pipelines)