Torna indietro e riesegui una pipeline

Importante

Questa funzionalità è in versione beta. Per abilitarla, imposta la pipeline sul canale Anteprima e imposta la configurazione pipelines.rewind.betaEnabled della pipeline su true. Vedere Requisiti.

Una trasformazione sbagliata, un lotto di record sorgente malformato o un cambiamento imprevisto dello schema possono lasciare le tabelle di una pipeline sbagliate da un momento noto in poi. Rewind riporta la pipeline a un punto precedente al problema, così da poter distribuire una correzione e riprocessare solo i dati interessati.

Rewind ripristina in modo coordinato le versioni delle tabelle, gli offset di origine dello streaming e lo stato dell'operatore, così la riesecuzione non salta record né genera duplicati. Tre operazioni rielaborano i dati e risolvono diversi problemi:

• Rewind si usa per una pipeline ripristinabile che ha scritto dati errati a partire da un momento noto, ad esempio dopo una trasformazione errata, un input malformato o una distribuzione di codice errata. Ripristina i dati della tabella, gli offset della sorgente e lo stato dell'operatore a un punto precedente al problema e rielabora solo i dati interessati, mantenendo lo stato dell'operatore. I dati precedenti a quel punto restano intatti.
• Il refresh completo ricostruisce una tabella con tutti i dati di origine disponibili e ne elimina il contenuto attuale. Usalo per ricalcolare tutto da zero, o quando una modifica di codice non è compatibile con lo stato esistente.
• Il reset del checkpoint recupera una pipeline il cui checkpoint è invalido o corrotto, o bloccata da un cambiamento di codice incompatibile con checkpoint. Resetta il checkpoint e prosegue in avanti preservando il contenuto attuale della tabella. Rewind non può recuperare questi casi, perché non allenta le regole di compatibilità dello Structured Streaming.

Requisiti

Requisito	Detail
Channel	La pipeline deve trovarsi nel canale Anteprima. Vedi Configurare le pipeline.
Configuration	Imposta la pipelines.rewind.betaEnabled configurazione della pipeline su true, poi esegui la pipeline una volta. Ogni flusso diventa riavvolgibile solo dopo aver completato un aggiornamento con il viaggio nel tempo abilitato.
Modalità pipeline	Pipeline attivate e continue. La modalità in tempo reale non è supportata.
Sources	Tabelle Delta, tabelle di streaming, Kafka e Auto Loader.
Obiettivi	Tabelle streaming e viste materializzate.
Flows	Flussi in streaming e flussi di cattura dati a cambio automatico (CDC), inclusi bersagli SCD Tipo 1 e SCD Tipo 2. Sono supportate query con stato come aggregazioni, join e deduplicazione.

Ogni flusso nella conduttura deve soddisfare questi requisiti. Mentre pipelines.rewind.betaEnabled è true, una pipeline che contiene un flusso che non soddisfa i requisiti non riesce ad aggiornarsi. Conferma che ogni flusso soddisfi i requisiti sopra prima di attivarlo.

Note

Una pipeline con pipelines.rewind.betaEnabled impostato su true non può tornare al canale Current finché il canale Current non viene aggiornato a un runtime che supporta la reversione.

Come funzionano il riavvolgimento e il rigioco

Riavvolgere e riprodurre sono passaggi separati.

Rewind ripristina ogni tabella alla versione che aveva in corrispondenza di un punto di riavvolgimento e reimposta i checkpoint di streaming che tengono traccia di fino a che punto ogni flusso ha letto. Le tue trasformazioni non vengono eseguite e nessun dato sorgente viene rielaborato.

Replay avviene alla successiva esecuzione della pipeline. Riprocessa dal punto di riavvolgimento usando la definizione attuale della pipeline, si riallinea al presente, quindi riprende la normale elaborazione incrementale. Rewind non avvia la pipeline, quindi avviala manualmente quando sei pronto.

La pipeline genera automaticamente i punti di riavvolgimento, circa una volta all'ora, e li mantiene per 7 giorni. Una pipeline appena creata non ha nulla a cui tornare indietro finché non produce il suo primo risultato.

Il riavvolgimento di un dataset riavvolge anche tutto ciò che è a valle all’interno della stessa pipeline. Rewind copre una sola pipeline. Non coordina con altre pipeline o con lettori esterni delle stesse tabelle, quindi gestisci queste separatamente.

Riavvolgi una pipeline usando l'interfaccia utente

La UI e Genie sono i principali modi per utilizzare Rewind. L'interfaccia indica i punti di riavvolgimento disponibili e mostra quali dataset ciascuno di essi influenza prima di effettuare il commit.

1. Nella pagina della tua pipeline, clicca accanto a Run pipeline, poi clicca su Rewind pipeline.
2. Seleziona un punto di riavvolgimento, oppure usa una scorciatoia come Riavvolgi fino a ieri o Riavvolgi fino all'ultimo punto. Fare clic su Avanti.
3. Seleziona quali tabelle includere. Usa la vista Grafico per selezionare i dataset nel grafo della pipeline, oppure la vista Lista per selezionarli da una tabella. Lascia selezionato Reimposta tutti i checkpoint (opzione predefinita) per ripristinare gli offset della sorgente e lo stato dell'operatore insieme ai dati della tabella, in modo che la pipeline elabori nuovamente dal punto di riavvolgimento. Cancella per ripristinare solo i dati della tabella, senza rielaborazione, ad esempio quando vuoi ripristinare il contenuto di una tabella ma non vuoi rielaborare i dati interessati. Questa impostazione deve essere la stessa per una tabella e per le relative tabelle upstream e non può essere disattivata per un flusso che legge un'origine esterna come Kafka o Auto Loader. Fare clic su Avanti.
4. Rivedi il punto di riavvolgimento, l'impostazione del checkpoint e i dataset interessati, poi clicca su Riavvolgimento.

Avvia la pipeline per riprodurre i dati.

Puoi riavvolgere ripetutamente. Ogni riavvolgimento sostituisce quello precedente, così puoi rimediare a una riproduzione non riuscita riavvolgendo altrove.

Riavvolgi una pipeline usando API, CLI o SDK

Oltre all'interfaccia utente e a Genie, puoi attivare un riavvolgimento programmatico con la CLI Databricks, l'API REST o l'SDK Python. Usa il percorso programmatico per l'automazione, come un processo di correzione, un harness di test o un orchestratore.

Un rollback è un singolo aggiornamento della pipeline il cui corpo della richiesta contiene un rewind_spec. I campi in rewind_spec impostano le stesse scelte che fai nell'interfaccia: il punto a cui riavvolgere, i dataset da includere e se resettare i checkpoint. Replay è il prossimo aggiornamento che invii senza rewind_spec, che è l'equivalente API di avviare la pipeline. Rewind non avvia la pipeline per te.

Puoi selezionare uno dei seguenti:

• Un punto di riavvolgimento (consigliato). Imposta rewind_point_id su uno dei punti di riavvolgimento generati dalla pipeline. Questa è la stessa scelta che fanno l'interfaccia utente e Genie. Il rewind point registra già la versione della tabella e il checkpoint da ripristinare per ogni dataset, quindi il rewind ripristina esattamente quello stato. Consulta Trovare un punto di ripristino.
• Un timestamp. Imposta rewind_timestamp a un punto arbitrario nel tempo in cui nessun punto di riavvolgimento generato corrisponde allo stato che vuoi. Scegli tu stesso il timestamp. Vedi Riavvolgere fino a una marca temporale.

Trova un punto di riavvolgimento

La pipeline registra ogni punto di riavvolgimento come rewind_summary evento nel suo registro eventi. Il/La id dell'evento è il/la rewind_point_id. Ogni punto periodico di riavvolgimento registra la versione corrente di ogni tabella nella pipeline. Un'esecuzione di prova registra anche un punto di riavvolgimento, dello stato che ha convalidato. Per elencare i punti di riavvolgimento degli ultimi 7 giorni, esegui una query come la seguente:

SELECT id AS rewind_point_id, timestamp, message
FROM event_log('<pipeline-id>')
WHERE event_type = 'rewind_summary'
  AND details:rewind_summary.dry_run = true
  AND timestamp > current_timestamp() - INTERVAL 7 DAYS
ORDER BY timestamp DESC;

I punti di riavvolgimento periodico riportano il messaggio Periodic rewind summary capturing current table versions.: Per annullare i dati errati, scegli il punto di riavvolgimento più recente il cui timestamp precede il momento in cui i dati errati sono stati scritti.

Se l'ID non corrisponde a un rewind_summary evento nel registro eventi della pipeline, il riavvolgimento fallisce con INVALID_REWIND_SPECIFICATION.REWIND_POINT_NOT_FOUND.

Corpo della richiesta di riavvolgimento

Per riavvolgere una pipeline, invia un aggiornamento con un corpo rewind_spec come il seguente:

Punto di riavvolgimento

{
  "cause": "USER_ACTION",
  "rewind_spec": {
    "rewind_point_id": "<rewind-point-id>",
    "datasets": [
      {
        "identifier": "catalog.schema.table",
        "cascade": true,
        "reset_checkpoints": true
      }
    ],
    "dry_run": false
  }
}

Timestamp

{
  "cause": "USER_ACTION",
  "rewind_spec": {
    "rewind_timestamp": "2026-09-29 08:36:55",
    "datasets": [
      {
        "identifier": "catalog.schema.table",
        "cascade": true,
        "reset_checkpoints": true
      }
    ],
    "dry_run": false
  }
}

Riavvolgi con il CLI

La CLI Databricks inoltra il corpo della richiesta tramite --json. Per riavvolgere una pipeline, esegui quanto segue:

Punto di riavvolgimento

Il comando databricks pipelines start-update elimina rewind_point_id dal corpo della richiesta. Invia invece la richiesta con databricks api post, che inoltra il corpo senza modificarlo:

databricks api post /api/2.0/pipelines/<pipeline-id>/updates --json @rewind.json --profile <profile>

Timestamp

databricks pipelines start-update <pipeline-id> --json @rewind.json --profile <profile>

Per riprodurre i dati, avvia un aggiornamento normale senza rewind_spec:

databricks pipelines start-update <pipeline-id> --profile <profile>

Per interrogare un aggiornamento fino a raggiungere uno stato terminale, esegui quanto segue:

databricks pipelines get-update <pipeline-id> <update-id> --profile <profile>

Riavvolgi con l'API REST

Per ripristinare una pipeline, invia una richiesta POST all'endpoint Start an update con il corpo rewind_spec. Ad esempio:

Punto di riavvolgimento

curl -sS -X POST \
  "https://<workspace-host>/api/2.0/pipelines/<pipeline-id>/updates" \
  -H "Authorization: Bearer $DATABRICKS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "cause": "USER_ACTION",
    "rewind_spec": {
      "rewind_point_id": "<rewind-point-id>",
      "datasets": [{"identifier": "catalog.schema.table", "cascade": true, "reset_checkpoints": true}],
      "dry_run": false
    }
  }'

Timestamp

curl -sS -X POST \
  "https://<workspace-host>/api/2.0/pipelines/<pipeline-id>/updates" \
  -H "Authorization: Bearer $DATABRICKS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "cause": "USER_ACTION",
    "rewind_spec": {
      "rewind_timestamp": "2026-09-29 08:36:55",
      "datasets": [{"identifier": "catalog.schema.table", "cascade": true, "reset_checkpoints": true}],
      "dry_run": false
    }
  }'

Torna indietro con l’SDK Python

Punto di riavvolgimento

L'helper tipizzato RewindSpec non include rewind_point_id. Invia la richiesta con il client REST autenticato dell'SDK invece. Questo funziona in ogni versione SDK, inclusa quella databricks-sdk inclusa con Databricks Runtime e computazione serverless. Per riavvolgere una pipeline, esegui quanto segue:

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()  # In a notebook, authenticates as the notebook identity.
resp = w.api_client.do("POST", f"/api/2.0/pipelines/{pipeline_id}/updates", body={
    "cause": "USER_ACTION",
    "rewind_spec": {
        "rewind_point_id": rewind_point_id,
        "datasets": [{"identifier": root_table, "cascade": True, "reset_checkpoints": True}],
        "dry_run": False,
    },
})
print(resp["update_id"])

Timestamp

L'helper tipizzato RewindSpec viene mappato direttamente al corpo della richiesta. Richiede databricks-sdk 0,40 e versioni successive. Per riavvolgere una pipeline, esegui quanto segue:

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.pipelines import StartUpdateCause, RewindSpec, RewindDatasetSpec

w = WorkspaceClient(profile="<profile>")
resp = w.pipelines.start_update(
    "<pipeline-id>",
    cause=StartUpdateCause.USER_ACTION,
    rewind_spec=RewindSpec(
        rewind_timestamp="2026-09-29 08:36:55",
        datasets=[RewindDatasetSpec(
            identifier="catalog.schema.table",
            cascade=True,
            reset_checkpoints=True,
        )],
        dry_run=False,
    ),
)
print(resp.update_id)

Il databricks-sdk in dotazione con Databricks Runtime, incluso il calcolo serverless, potrebbe precedere il RewindSpec tipizzato. In un notebook/Jupyter notebook, usa il client REST autenticato, disponibile in ogni versione dell'SDK, così da non dover installare un SDK più recente:

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()  # Authenticates as the notebook identity.
w.api_client.do("POST", f"/api/2.0/pipelines/{pipeline_id}/updates", body={
    "cause": "USER_ACTION",
    "rewind_spec": {
        "rewind_timestamp": rewind_timestamp,
        "datasets": [{"identifier": root_table, "cascade": True, "reset_checkpoints": True}],
        "dry_run": False,
    },
})

Riavvolgi il riferimento della specifica

La tabella seguente descrive i campi nel corpo rewind_spec.

Field	Description
rewind_point_id	Il id di un evento rewind_summary a cui ripristinare. Consulta Trova un punto di ripristino. Imposta esattamente uno di rewind_point_id o rewind_timestamp.
rewind_timestamp	Il momento a cui ripristinare, in formato UTC yyyy-MM-dd HH:mm:ss. Vedi Riavvolgi a un timestamp. Imposta esattamente uno di rewind_point_id o rewind_timestamp.
datasets	L'elenco dei dataset da reimpostare. Ometti la lista per riavvolgere tutti i dataset nella pipeline, ciascuno con cascade e reset_checkpoints impostati su true.
datasets[].identifier	Il nome completamente qualificato del dataset da riavvolgere, in forma catalog.schema.table.
datasets[].cascade	Se riavvolgere anche ogni set di dati a valle di questo nella pipeline. Specifica questo campo per ogni dataset. Per una pipeline lineare, impostala su true sul dataset radice per reimpostare l'intera pipeline.
datasets[].reset_checkpoints	Se reimpostare gli offset della sorgente in streaming e lo stato dell'operatore in modo che la riesecuzione riparta dalla rielaborazione dal punto di riavvolgimento. Questa è l'impostazione Reimposta tutti i checkpoint dell'interfaccia utente. Di default è true. Impostalo su false per ripristinare solo i dati delle tabelle, senza rielaborazione.
dry_run	Se validare il rewind senza cambiare alcuna tabella. Di default è false. Vedi Esegui una prova di riavvolgimento con una simulazione.

Warning

Quando un dataset root legge una sorgente esterna come Kafka o Auto Loader, si mantiene reset_checkpoints impostato su true. L'interfaccia impedisce di azzerare Reset tutti i checkpoint per queste sorgenti, ma l'API accetta una richiesta che imposta reset_checkpoints a false. La richiesta viene accettata e il rewind può iniziare, ma la pipeline fallisce dopo. Limita reset_checkpoints: false ai dataset che un flusso in streaming non legge.

Riavvolgi fino a un timestamp

Usa rewind_timestamp quando nessun punto di riavvolgimento generato corrisponde allo stato che vuoi, ad esempio per riavvolgere fino a un punto tra due punti di riavvolgimento. Fornisci tu stesso il timestamp. Segui due regole:

• Scegli un punto strettamente tra due commit, non uno che ricade su un commit. Un timestamp che corrisponde a un commit può causare un errore di replay. Usa il punto intermedio tra l'ultimo commit buono e il primo commit cattivo.
• Su una pipeline che non è stata riavvolta o ricaricata, scansiona il feed di dati delle modifiche a partire dalla versione 1. La versione 0 non registra i dati di modifica. Se la versione 1 è più vecchia del periodo di conservazione dei file della tabella (7 giorni per impostazione predefinita), la scansione fallisce, quindi inizia dalla versione più vecchia che è stata registrata entro quel periodo.

La query qui sotto, che trova i tempi di commit, legge il feed dei dati di modifica (CDF) della tabella della pipeline che vuoi riavvolgere, quindi il CDF deve essere abilitato su quella tabella. Il riavvolgimento in sé non richiede CDF. Le pipeline Lakeflow abilitano di default il CDF sulle loro tabelle, quindi la query funziona senza una configurazione aggiuntiva. Se il CDF è stato disabilitato sulla tabella, usa un punto di riavvolgimento oppure usa DESCRIBE HISTORY per scegliere un punto.

Abbina il filtro _change_type all'errore che stai indagando: per inserimenti o aggiornamenti difettosi, tieni 'insert' e 'update_postimage'; per una cancellazione errata, filtra su 'delete' invece, perché table_changes mostra solo le righe eliminate sotto quel tipo di modifica. Per calcolare un timestamp del punto medio da un predicato che seleziona i dati errati, esegui una query come segue:

WITH onset AS ( -- The first commit that introduced the bad data.
  SELECT min(_commit_version) AS first_bad_version,
         min(_commit_timestamp) AS first_bad_ts
  FROM table_changes('catalog.schema.table', 1) -- Start from version 1.
  WHERE _change_type IN ('insert', 'update_postimage') -- Use ('delete') instead to find an erroneous delete.
    AND (event_date >= '2026-09-29') -- Example predicate. Replace the condition in parentheses with one that selects the bad rows.
),
bounds AS (
  SELECT first_bad_ts,
         (SELECT max(h.timestamp)
          FROM (DESCRIBE HISTORY catalog.schema.table) h
          WHERE h.version < onset.first_bad_version) AS last_good_ts
  FROM onset
)
SELECT date_format(
         from_unixtime((unix_timestamp(last_good_ts) + unix_timestamp(first_bad_ts)) / 2),
         'yyyy-MM-dd HH:mm:ss') AS rewind_timestamp
FROM bounds;

Se la pipeline è già stata riavvolta o aggiornata, avviare la scansione dall'ultimo RESTORE o dal commit di refresh in modo che non superi l'operazione precedente.

Esegui un riavvolgimento con una simulazione

Per convalidare un rewind senza cambiare alcuna tabella, imposta dry_run su true. L'aggiornamento passa attraverso la convalida fino allo stato terminale COMPLETED senza modificare i dati. Usa una prova a vuoto per confermare che la pipeline accetta il ripristino, poi invia di nuovo la richiesta impostando dry_run a false.

Un'esecuzione di prova registra il rewind che è stato convalidato come evento rewind_summary. Per applicare esattamente quel riavvolgimento, ad esempio dopo una prova a vuoto con rewind_timestamp, invia il riavvolgimento reale con il id dell'evento come rewind_point_id. Un’esecuzione di prova che fallisce non blocca gli aggiornamenti successivi.

Quando esegui cicli successivi di riavvolgimento e riproduzione in automazione, esegui un refresh completo tra di essi. Sovrapporre un secondo riavvolgimento e una riproduzione su una pipeline senza un aggiornamento completo nel mezzo può duplicare le righe rielaborate nelle tabelle di streaming a valle.

Dopo un riavvolgimento

Un guasto alla ripresa lascia la condotta riavvolta ma fermata. Correggi il codice o i dati sorgente e avvia la pipeline per riprovare, oppure riavvolgi a un punto diverso. La pipeline non esegue automaticamente il rollback e gli errori vengono segnalati tramite la diagnostica standard della pipeline e il registro eventi.

La riesecuzione ha esito positivo solo se la definizione corrente della pipeline è compatibile con lo stato ripristinato; il riavvolgimento non deroga alle regole di compatibilità di Structured Streaming. Per quali modifiche sono compatibili, vedi Tipi di modifiche nelle query di Structured Streaming. Le viste materializzate seguono la semantica batch e tollerano cambiamenti più ampi dello schema, ma falliscono comunque se una dipendenza è incompatibile.

Un riavvolgimento che non va a buon fine a metà percorso può lasciare la pipeline parzialmente riavvolta. È possibile procedere in due modi:

1. Riavvolgendo nuovamente fino allo stesso punto o a un punto diverso, la pipeline converge in quel punto.
2. Per forzare la pipeline ad avviare un aggiornamento normale nonostante il riavvolgimento incompleto, imposta pipelines.allowUpdateAfterIncompleteRewind su true e riavvia la pipeline.

Note

Se un rewind fallisce o lo invii con un punto di rewind non valido, la pipeline blocca ogni aggiornamento successivo non di rewind, incluso un aggiornamento completo, finché non invii nuovamente un rewind valido. Quando attivi i riavvolgimenti tramite codice, conferma che rewind_point_id o rewind_timestamp si riferisca a un punto valido prima di inviare la richiesta.

Quanto indietro puoi riavvolgere

I punti di riavvolgimento sono conservati per 7 giorni. In quella finestra, un riavvolgimento non riesce se i dati di cui ha bisogno sono già stati rimossi. Controlla questi prima di riavvolgere:

• VACUUM O un cortometraggio delta.deletedFileRetentionDuration sui tuoi tavoli. Consulta Lavora con la cronologia delle tabelle.
• Un periodo di conservazione dei dati di origine più breve della finestra temporale a cui vuoi tornare indietro, ad esempio un topic Kafka che conserva i dati per un giorno.

Le pipeline con più sorgenti o con lunghe catene di dipendenze necessitano di un periodo di conservazione più lungo, perché ogni tabella e ogni punto di controllo devono poter risalire fino a un punto coerente.

Limitations

• Rewind non può ripristinare una pipeline a uno stato precedente a un aggiornamento completo.
• I punti di ripristino vengono conservati per 7 giorni e il ripristino non riesce se la cronologia della tabella o i dati di origine necessari per tornare a quel punto sono già stati rimossi, ad esempio da VACUUM o da un breve periodo di conservazione dei dati di origine. Vedi quanto puoi riavvolgere indietro.
• La modalità in tempo reale non è supportata.
• Kinesis, Pulsar, Google Pub/Sub e sorgenti personalizzate costruite con le API di sorgenti dati DSv2 o Python non sono supportate come sorgenti.
• I dissipatori esterni e i dissipatori personalizzati non sono supportati, inclusi i dissipatori definiti con create_sink(). Vedere Usare sink nelle pipeline.
• Le tabelle di streaming che utilizzano un filtro a riga o una maschera a colonna non possono essere riavvolte. Vedere Applicare manualmente filtri di riga e maschere di colonna.
• Il rewind con stato richiede l'archivio di stato RocksDB, utilizzato dalle pipeline per impostazione predefinita. Un riavvio non riesce per un flusso configurato con un archivio di stato diverso.
• Alcune tabelle di streaming AUTO CDC necessitano di un aggiornamento prima di poter essere riportate indietro. La pipeline ti avvisa quando richiedi di riavvolgerla.
• Le viste materializzate potrebbero essere ricalcolate completamente anziché essere aggiornate in modo incrementale dopo un riavvolgimento. Vedere aggiornamento incrementale per le viste materializzate.
• Rewind si applica a una pipeline e non si coordina con lettori esterni o con altre pipeline che leggono le stesse tabelle.
• Rewind non ripristina il codice della pipeline, la configurazione della pipeline o i metadati degli oggetti del Catalogo Unity come tag e concessioni.

Risorse aggiuntive

• Ripristinare una pipeline da un guasto del checkpoint di streaming
• Aggiornamento completo per le tabelle di streaming
• Come si aggiornano le pipeline?
• Eseguire un aggiornamento della pipeline
• Usare la cronologia delle tabelle