Aanvullen van historische gegevens met datapijplijnen

In data engineering verwijst backfilling naar het proces van retroactief verwerken van historische gegevens via een gegevenspijplijn die is ontworpen voor het verwerken van huidige of streaminggegevens.

Normaal gesproken is dit een afzonderlijke stroom die gegevens naar uw bestaande tabellen verzendt. In de volgende afbeelding ziet u een backfillstroom die historische gegevens naar de bronstabellen in uw pijplijn verzendt.

Voor sommige scenario's waarvoor mogelijk een backfill is vereist:

• Historische gegevens van een verouderd systeem verwerken om een machine learning-model (ML) te trainen of een historisch trendanalysedashboard te bouwen.
• Hiermee wordt een subset van gegevens opnieuw verwerkt vanwege een probleem met gegevenskwaliteit met upstream-gegevensbronnen.
• Uw zakelijke vereisten zijn gewijzigd en u moet gegevens invullen voor een andere periode die niet wordt gedekt door de eerste pijplijn.
• Uw bedrijfslogica is gewijzigd en u moet zowel historische als huidige gegevens opnieuw verwerken.

De backfillflow die je gebruikt hangt af van de doeltabel en brondata. Voor een AUTO CDC langzaam veranderende dimensie (SCD) Type 1 target met een gezaghebbende snapshot, gebruik een eenmalige AUTO CDC FROM SNAPSHOT flow. Voor een SCD-migratie die historische wijzigingen opnieuw afspeelt, gebruik een eenmalige AUTO CDC stroom.

Voor alleen append-backfills: Gebruik een gespecialiseerde appendflow met de ONCE optie om een append-only streamingtabel te backfillen. Zie append_flow of CREATE FLOW (pijplijnen) voor meer informatie over de ONCE optie.

Overwegingen bij het doorvoeren van historische gegevens in een streamingtabel

• Voeg doorgaans de gegevens toe aan de bronsstreamingtabel. De downstream zilver- en goudlagen pikken de nieuwe gegevens van de bronzen laag op.
• Zorg ervoor dat uw pijplijn dubbele gegevens probleemloos kan verwerken als dezelfde gegevens meerdere keren worden toegevoegd.
• Zorg ervoor dat het historische gegevensschema compatibel is met het huidige gegevensschema.
• Houd rekening met de datavolumegrootte en de vereiste processing service-level agreement (SLA), en configureer dienovereenkomstig de cluster- en batchgroottes.

Voorbeeld: Voeg een backfill toe aan een bestaande pipeline

In dit voorbeeld hebt u een pijplijn die onbewerkte gebeurtenisregistratiegegevens opneemt uit een cloudopslagbron vanaf 01 januari 2025. Later realiseer je je dat je de historische data van de vorige drie jaar met terugwerkende kracht wilt laden voor downstream rapportage- en analysetoepassingen. Alle gegevens bevinden zich op één locatie, gepartitioneerd op jaar, maand en dag, in JSON-indeling.

Eerste pijplijn

Hier volgt de beginnende pijplijncode die incrementeel de onbewerkte gebeurtenisregistratiegegevens uit de cloudopslag opneemt.

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 gebruiken we de modifiedAfter Auto Loader-optie om ervoor te zorgen dat we niet alle gegevens verwerken van het cloud-opslagpad. De incrementele verwerking wordt bij die grens afgekapt.

Backfill-gegevens uit de afgelopen 3 jaar

Nu wilt u een of meer stromen toevoegen om eerdere gegevens in te vullen. Voer in dit voorbeeld de volgende stappen uit:

• Gebruik de append once workflow. Hiermee wordt een eenmalige backfill uitgevoerd zonder verder te gaan na die eerste backfill. De code blijft aanwezig in uw pijplijn en als de pijplijn ooit volledig wordt vernieuwd, wordt de backfill opnieuw uitgevoerd.
• Maak drie backfill-stromen, één voor elk jaar (in dit geval worden de gegevens in het pad gesplitst per jaar). Voor Python parameteriseren we het maken van de stromen, maar in SQL herhalen we de code drie keer, één keer voor elke stroom.

Als u aan uw eigen project werkt en geen serverloze berekeningen gebruikt, kunt u de maximale werkrollen voor de pijplijn bijwerken. Het verhogen van het maximum aantal werknemers zorgt ervoor dat u over de middelen beschikt om de historische gegevens te verwerken, terwijl u doorgaat met het verwerken van de huidige streaminggegevens binnen de verwachte SLA.

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

Deze implementatie markeert verschillende belangrijke patronen.

Scheiding van verantwoordelijkheden

• Incrementele verwerking is onafhankelijk van backfill-bewerkingen.
• Elke stroom heeft zijn eigen configuratie- en optimalisatie-instellingen.
• Er is een duidelijk onderscheid tussen incrementele en backfillbewerkingen.

Gecontroleerde uitvoering

• Als u de ONCE optie gebruikt, zorgt u ervoor dat elke backfill precies eenmaal wordt uitgevoerd.
• De backfillstroom blijft in de pijplijngrafiek staan, maar wordt inactief nadat deze is voltooid. Het is automatisch gereed voor gebruik bij een volledige verversing.
• Er is een duidelijk audit-trail van backfill-operaties in de definitie van de pijplijn.

Optimalisatie van verwerking

• U kunt de grote backfill splitsen in meerdere kleinere backfills voor snellere verwerking of voor controle over de verwerking.
• Met verbeterde automatische schaalaanpassing wordt de clustergrootte dynamisch geschaald op basis van de huidige clusterbelasting.

Ontwikkeling van schema's

• Het gebruik van schemaEvolutionMode="addNewColumns" gaat soepel om met schemawijzigingen.
• U hebt consistente schemadeductie voor historische en huidige gegevens.
• Er is een veilige verwerking van nieuwe kolommen in nieuwere gegevens.

Voeg een backfill toe aan een AUTO CDC SCD Type 1 tabel

Gebruik een eenmalige AUTO CDC FROM SNAPSHOT flow om een gezaghebbende snapshot toe te voegen aan een SCD Type 1-doel dat ook een lopende change data capture (CDC)-feed ontvangt. De snapshotversie en de CDC-sequencingkolom vormen één ordeningsdomein. Een nieuwere CDC-gebeurtenis gaat boven een oudere snapshot, terwijl een nieuwere snapshot voorrang krijgt boven een oudere CDC-gebeurtenis.

Requirements

Voordat je de backfill toevoegt, zorg ervoor dat de flows aan de volgende eisen voldoen:

• Het doel gebruikt SCD Type 1.
• Het doel heeft precies één AUTO CDC FROM SNAPSHOT flow en een of meer uniek benoemde AUTO CDC flows.
• Alle flows gebruiken hetzelfde aantal keys in dezelfde volgorde. Snapshot-flow sleutelnamen worden zonder onderscheid tussen hoofdletters en kleine letters vergeleken met AUTO CDC sleutelnamen. Meerdere stromen AUTO CDC moeten identieke sleutelnamen en hoofdlettergebruik gebruiken.
• De snapshotversie en elke CDC-sequencingkolom hebben exact hetzelfde datatype.
• De AUTO CDC FROM SNAPSHOT flow definieert geen verwachtingen.
• De stromen AUTO CDC gebruiken IGNORE NULL UPDATES niet. In Python moet je ignore_null_updates_column_list, ignore_null_updates_except_column_list of ignore_null_updates niet instellen.
• De pijplijn gebruikt de getriggerde modus. Dit patroon ondersteunt geen continue pipelines.

Beide flowtypes kunnen de SQL- of Python-pijplijninterface gebruiken. Je kunt SQL- en Python-flows combineren in dezelfde target.

De snapshot moet de volledige staat van de bron in zijn versie weergeven. Als er een doel-sleutel ontbreekt in de snapshot, behandelt AUTO CDC FROM SNAPSHOT de afwezigheid bij de snapshotversie als een verwijdering. Een CDC-gebeurtenis met een nieuwere versie bewaart of herstelt de sleutel.

Voeg de opvulling toe

Om een eenmalige snapshot-backfill toe te voegen en door te gaan met het verwerken van CDC-gebeurtenissen, gebruik je de volgende stappen:

1. Houd de bestaande doeltabel en de lopende AUTO CDC-stromen in de pijplijndefinitie.
2. Definieer de canonieke snapshot en de bijbehorende versie. Voor een Python-callback moet de callback bij de eerste aanroep een snapshot en versie teruggeven. Pas terug None nadat ten minste één snapshot is verwerkt.
3. Voeg één AUTO CDC FROM SNAPSHOT flow toe met once=True in Python of ONCE in SQL. Voor een SQL-backfill in een bestaand doel, voeg een WITH VERSION query toe. Een SQL-snapshotflow zonder WITH VERSION ondersteunt alleen een eerste lading in een leeg doel.
4. Voer een getriggerde pijplijnupdate uit om de backfill en lopende CDC-gebeurtenissen te verwerken.

Het volgende voorbeeld begint met een bestaande Python-pijplijn die wijzigingen incrementeel verwerkt vanuit customers_cdc. Stel dat je deze pipeline al hebt uitgevoerd en de customers-target hebt gevuld:

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

Om dat bestaande doel met terugwerkende kracht aan te vullen met de staat van customers_snapshot per 1 januari 2025, voeg de volgende code toe aan dezelfde pijpleidingdefinitie. Houd de bestaande doeltabel en AUTO CDC flow aan:

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

De callback moet bij de eerste aanroep een snapshot en een versie teruggeven. Als het terugkeert None voordat een snapshot is verwerkt, mislukt de pipeline-update. Nadat de snapshot is verwerkt, duidt het retourneren van None erop dat er geen extra snapshots beschikbaar zijn.

De snapshotversie is een Python literaldatetime, wat overeenkomt met het Spark SQL-type LiteralTypeTIMESTAMP. De bestaande AUTO CDC flow cast change_timestamp naar TIMESTAMP, zodat de twee sequencingtypes exact overeenkomen. Het voorbeeld gebruikt Python voor beide stromen, maar je kunt een van beide stromen definiëren in SQL en SQL- en Python-stromen in hetzelfde doel mengen. Voor SQL-syntaxis, inclusief de vereiste WITH VERSION query voor een niet-leeg doel, zie CREATE FLOW (pipelines).

Nadat de snapshot-flow succesvol is gecommit, slaan volgende incrementele updates deze over terwijl de AUTO CDC flow doorgaat met het verwerken van nieuwe gebeurtenissen.

Dit uniforme backfill-patroon ondersteunt geen SCD Type 2 of bitemporele doeltabellen.

Voorbeeld: Een SCD-doel opvullen tijdens een migratie

Een veelvoorkomend migratiescenario is een tabel met langzaam veranderende dimensie (SCD) die al bestaat in een legacy-systeem met jaren aan opgebouwde geschiedenis, maar waarvan de oorspronkelijke wijzigingsfeed niet langer beschikbaar is. Omdat de oorspronkelijke wijzigingsgebeurtenissen verdwenen zijn, speel je in plaats daarvan de geschiedenis van de legacy-tabel één keer af in het nieuwe AUTO CDC doel, waarna je vanaf dan een nieuwe CDC-feed toevoegt. Voor meer informatie over AUTO CDC en SCD-types, zie De AUTO CDC-API's: het vastleggen van gegevenswijzigingen vereenvoudigen met pijplijnen.

Het patroon is een eenmalige AUTO CDC gegevensstroom naar dezelfde streamingtabel waarop de doorlopende AUTO CDC gegevensstroom is gericht. Een AUTO CDC doel accepteert alleen AUTO CDC flows, dus de seed moet ook een AUTO CDC flow zijn. Een eenvoudige INSERT INTO ONCE append-stroom naar dezelfde tabel mislukt de validatie:

1. Maak de doelstroomtabel aan waar je AUTO CDC flow naartoe schrijft.
2. Seed de legacy-geschiedenis één keer met een AUTO CDC ONCE flow die de legacy SCD-tabel als een stream leest, gesorteerd op de startkolom van legacy-validiteit. Speel de legacy-rijen opnieuw af als veranderingsgebeurtenissen in plaats van ze zelf vorm te geven. AUTO CDC bouwt de __START_AT en __END_AT geschiedeniskolommen voor een SCD Type 2 doel, dus schrijf die kolommen niet direct.
3. Voeg de lopende AUTO CDC flow toe die de nieuwe wijzigingenfeed leest. AUTO CDC hanteert de volgorde per sleutel, dus de overschakeling moet voor elke bedrijfssleutel afzonderlijk worden aangehouden: de eerste live wijziging van elke sleutel moet volgen op de laatste vooraf geladen wijziging voor diezelfde sleutel. Een sequentiewaarde die net later ligt dan het globale legacy-maximum kan nog steeds verouderd zijn voor een individuele sleutel, en de eerste live-wijziging van die sleutel wordt dan genegeerd of verkeerd gerangschikt.

De volgende code maakt een streamingtabel die de bovenstaande stappen gebruikt:

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 stromen moeten het eens zijn over hun sleutels, hun SCD-type en het datatype van hun sequentiekolom. In het voorgaande voorbeeld worden beide stromen geordend op basis van een tijdstempel, waarbij één cutovertijd wordt gebruikt om de vooraf geladen geschiedenis van de livegegevensstroom te scheiden. Als de legacy-tabel een andere waarde volgt dan de live-feed, cast dan één ervan zodat de types overeenkomen.

Dezelfde opzet werkt voor een SCD Type 1-doel: verander STORED AS SCD TYPE 2 in STORED AS SCD TYPE 1 in beide stromen, en het doel behoudt alleen de huidige rij per sleutel. Voordat je op een van beide vormen vertrouwt, valideer je op een steekproef van sleutels dat de eerste live wijziging voor een gezaadde sleutel precies één nieuwe versie oplevert en de vorige correct sluit. Bij die stap treedt meestal een sequentiëringsgat per toets op.

Aanvullende bronnen

• Gegevens incrementeel laden en verwerken met Lakeflow-pijplijnstromen
• Stromen gebruiken in Lakeflow-pijplijnen
• De AUTO CDC-API's: het vastleggen van wijzigingsgegevens vereenvoudigen met pijplijnen
• append_flow
• create_auto_cdc_from_snapshot_flow
• create_auto_cdc_flow
• CREATE FLOW (pijplijnen)