Garanzie di lavorazione nelle pipeline Lakeflow

Nuovi tentativi e riesecuzioni sono inevitabili in qualsiasi pipeline reale, quindi questa pagina spiega le garanzie di elaborazione offerte dalle pipeline Lakeflow e come fare in modo che le parti che scrivi possano essere rieseguite in sicurezza.

Overview

Due proprietà correlate determinano se la riesecuzione di una pipeline sia sicura:

  • Idempotenza significa che una pipeline produce lo stesso risultato indipendentemente da quante volte la esegui sullo stesso input. Ripetere l’esecuzione dopo un errore, eseguire due volte il backfill di un intervallo di date o riattivare manualmente un processo non creano mai righe duplicate né corrompono lo stato.
  • La garanzia di elaborazione descrive quante volte ogni record influisce sul risultato. Almeno una volta l'elaborazione garantisce che ogni record venga elaborato, ma un fallimento e un nuovo tentativo potrebbero processare alcuni record più di una volta, rischiando duplicati. L'elaborazione exactly-once garantisce che ogni record incida sul risultato come se fosse stato elaborato esattamente una sola volta, anche in caso di nuovi tentativi, senza duplicati e senza omissioni.

Le pipeline di Lakeflow sono idempotenti per impostazione predefinita per i componenti che gestiscono e forniscono un'elaborazione esattamente una volta nelle proprie tabelle gestite. La cosa importante da capire è dove queste garanzie smettono di essere automatiche, così puoi aggiungere le giuste misure di sicurezza ai margini della tua pipeline.

Come funziona

Le pipeline di Lakeflow forniscono elaborazione exactly-once e idempotenza per i flussi che gestiscono e offrono anche strumenti per fare in modo che la logica che scrivi sia idempotente.

Elaborazione una sola volta per le tabelle gestite

Nelle tabelle gestite, per impostazione predefinita si dispone dell'elaborazione una sola volta. Le tabelle di streaming utilizzano i checkpoint di Structured Streaming combinati con le scritture transazionali di Delta Lake: ogni micro-batch registra insieme i propri offset di origine e il relativo output, quindi un batch rieseguito dopo un errore o va a buon fine completamente oppure viene interamente annullato e rieseguito, senza mai essere applicato parzialmente due volte. Ciò vale per l'ingestione di file con Auto Loader, la lettura da Kafka, Kinesis e Hub eventi di Azure e AUTO CDC gli upsert, senza che tu debba scrivere codice.

Se almeno una volta un sorgente invia lo stesso record più volte, la pipeline li elabora come record unici e li scrive tutti nella tua tabella. Rimuovere quei duplicati è una tua responsabilità. Vedi Deduplicare almeno una volta le fonti.

L'idempotenza per le letture deriva dagli stessi checkpoint. Auto Loader e i checkpoint delle tabelle di streaming garantiscono che ogni file di origine o offset venga elaborato una sola volta ai fini del monitoraggio dello stato, quindi, se un aggiornamento della pipeline viene rielaborato dopo un guasto, l’elaborazione riparte dal checkpoint invece di rielaborare o saltare i dati. Lo ottieni usando tabelle di streaming su spark.readStream invece di loop batch scritti manualmente. Vedere Tabelle di streaming.

Usa AUTO CDC invece di MERGE scritto manualmente

AUTO CDC INTO è intrinsecamente idempotente rispetto al suo keys e sequence_by. Applicando due volte lo stesso record di modifica, o applicando record fuori ordine, si produce lo stesso stato finale, perché la pipeline utilizza la colonna di sequenza per decidere se una riga in ingresso è effettivamente più recente di quella memorizzata:

CREATE FLOW customers_cdc_flow AS AUTO CDC INTO customers_silver
FROM stream(customers_cdc_bronze)
KEYS (customer_id)
SEQUENCE BY sequence_num
STORED AS SCD TYPE 1;

Se scrivi una logica di upsert personalizzata al di fuori di AUTO CDC (cosa rara, ma a volte necessaria per condizioni di merge complesse), basala su una chiave di business stabile e fa' in modo che possa essere applicata due volte in sicurezza, ad esempio una MERGE ... WHEN MATCHED basata su order_id invece di una INSERT alla cieca. Per maggiori informazioni, vedi Le API AUTO CDC: Semplifica la cattura dei dati dei cambiamenti con pipeline.

Tieni le tue trasformazioni idempotenti

Per mantenere la logica idempotente durante la riesecuzione di operazioni di scrittura, segui queste due linee guida:

  • Evitare trasformazioni non deterministiche nelle viste materializzate. Poiché una vista materializzata può essere ricalcolata completamente o in modo incrementale, evita le funzioni il cui risultato dipende da quando vengono eseguite piuttosto che da quale sia l’input. Ad esempio, non usare current_timestamp() per calcolare un valore aziendale che dovrebbe rimanere fisso una volta scritto; prendi il timestamp dall'evento sorgente o invialo come parametro in modo che il ricalcolo produca output identico.
  • Progetta aggiornamenti completi per sicurezza. Un aggiornamento completo elimina e ricalcola una tabella da zero, ed è sicuro solo se ogni fonte a monte può ancora produrre lo storico completo. Se una fonte a monte espone solo una finestra mobile di modifiche, un aggiornamento completo di una tabella a valle AUTO CDC può comportare la perdita silenziosa della cronologia; quindi, progetta la conservazione della sorgente e dei topic tenendone conto.

Arriva esattamente una volta ai bordi

Il comportamento exactly-once cessa di essere automatico ai confini di ciò che la pipeline controlla direttamente, ad esempio nelle scritture verso sistemi esterni. Quando ti espandi verso un sistema esterno, rendi la scrittura stessa idempotente, ad esempio tramite l'upserting per chiave sul lato ricevente, dato che un micro-batch riprovato potrebbe altrimenti scrivere lo stesso batch due volte. Il seguente sink scrive ogni partizione del batch negli executor e utilizza una chiave di idempotenza affinché un batch rieseguito non venga scritto due volte:

from pyspark import pipelines as dp

@dp.foreach_batch_sink(name="orders_to_external_api")
def write_orders_to_api(batch_df, batch_id):
    def write_partition(rows):
        # Open one client per partition.
        for row in rows:
            # Use an idempotency key (order_id) so a retried batch doesn't double-write.
            upsert_to_external_system(key=row.order_id, payload=row.asDict())

    batch_df.select("order_id", "amount").foreachPartition(write_partition)

Per maggiori informazioni sulla scrittura su sistemi esterni, consulta Sinks in Lakeflow pipelines.

Deduplicare almeno una volta le fonti

Quando una sorgente può consegnare un record più di una volta, si deduplica a valle. Combina una watermark con dropDuplicatesWithinWatermark, che è compatibile con le watermark e non richiede uno stato illimitato per rilevare i duplicati. Rimuovi i duplicati in base alle colonne che identificano univocamente un evento. L'identità può coprire più colonne quando nessuna singola colonna è unica di per sé. Nel seguente esempio, un numero di sequenza di clic è unico solo all'interno della sua sessione, quindi le due colonne insieme identificano l'evento:

from pyspark import pipelines as dp

@dp.table(name="clicks_deduped")
def clicks_deduped():
    return (
        spark.readStream.table("clicks_bronze")
        .withWatermark("click_ts", "5 minutes")
        .dropDuplicatesWithinWatermark(["session_id", "click_seq_num"])
    )

Scegli quelle colonne dal contratto di unicità della fonte, non da ciò che appare distinto nei dati campione. Le colonne che possono legittimamente ripetere scartano eventi reali quando li consideri come l'identità. Un utente che clicca due volte sullo stesso annuncio è un esempio comune: la deduplicazione sull'utente e l'annuncio lascia silenziosamente cadere il secondo clic.

La semantica di upsert basata su chiavi di AUTO CDC elimina naturalmente i duplicati, quindi instradare dati con semantica at-least-once attraverso un flusso AUTO CDC basato su una chiave di business stabile è un altro modo per convergere su uno stato con semantica exactly-once.

Limitazioni

L'elaborazione eseguita una sola volta si applica ai flussi Delta-to-Delta gestiti. Considera le seguenti connessioni come operazioni con garanzia di esecuzione almeno una volta e aggiungi esplicitamente una logica di deduplicazione o di scrittura idempotente.

  • foreach_batch_sink e operazioni di scrittura esterne personalizzate. Spark garantisce che un batch venga tentato almeno una volta, ma un batch riprovato dopo una scrittura parziale può lasciare alcune righe visibili due volte nel sistema esterno. Rendere idempotente la scrittura esterna, ad esempio eseguendo un upsert su una chiave naturale oppure scrivendo un identificativo del batch che il destinatario possa usare per eliminare i duplicati.
  • Kafka come un lavandino. I temi di Kafka non supportano scritture transazionali esattamente una volta come fa Delta, quindi una scrittura micro-batch riprovata su Kafka può produrre messaggi duplicati. Se i consumatori a valle sono sensibili ai duplicati, dedupe dal lato del consumatore, ad esempio tramite l'ID dell'evento.
  • Fonti dati Python personalizzate utilizzate come sorgenti. Il fatto che le letture vengano elaborate una sola volta dipende dal fatto che l'implementazione della sorgente segnali correttamente gli offset e riprenda da essi. Se non traccia gli offset, consideralo come at-least-once e deduplica a valle con dropDuplicates su un ID evento oppure affidandoti alla semantica di upsert basata sulla chiave di AUTO CDC.

Come regola generale, se l'intera tua pipeline è Delta-to-Delta (tabelle in streaming e visualizzazioni materializzate che leggono e scrivono tabelle Delta tramite flussi gestiti), hai già esattamente una volta. Nel momento in cui aggiungi un foreach_batch_sink, un sink non-Delta, o una fonte personalizzata non verificata, considera quel bordo specifico almeno una volta e aggiungi lì la logica di scrittura di idempotenti o deduplicazione.

Risorse aggiuntive

  • Last updated on