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.

Een backfill in Lakeflow-pijplijnen wordt ondersteund met een gespecialiseerde toevoegstroom die gebruikmaakt van de ONCE optie. 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. Downstream zilver- en goudlagen halen de nieuwe gegevens uit de bronslaag 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 grootte van het gegevensvolume en de vereiste SLA voor verwerkingstijd en configureer het cluster en de batchgrootten dienovereenkomstig.

Voorbeeld: Een backfill toevoegen aan een bestaande pijplijn

In dit voorbeeld hebt u een pijplijn die onbewerkte gebeurtenisregistratiegegevens opneemt uit een cloudopslagbron vanaf 01 januari 2025. U realiseert zich later dat u de vorige drie jaar historische gegevens wilt invullen voor downstreamrapportage en analyse van gebruiksscenario's. 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.

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 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. Vul de bestaande geschiedenis één keer met een AUTO CDC ONCE-flow die de bestaande SCD-tabel als stream leest, gesorteerd op basis van de bestaande kolom voor het begin van de geldigheidsperiode. 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
• append_flow
• CREATE FLOW (pijplijnen)