Pular para o conteúdo principal
Última atualização em

Restaurar e reproduzir um pipeline

info

Beta

Este recurso está em Beta. To enable it, set your pipeline to the Preview canal and set the pipelines.rewind.betaEnabled pipeline configuration to true. See Requirements.

Uma transformação inadequada, um lote de registros de origem malformado ou uma alteração inesperada de esquema podem deixar as tabelas de um pipeline incorretas de um ponto conhecido no tempo em diante. A restauração retorna o pipeline para um ponto anterior ao problema para que você possa implantar uma correção e reprocessar apenas os dados afetados.

A retrocessão restaura versões de tabela, desvios de fonte de transmissão e estado do operador juntos, de modo que a repetição não ignore registros nem grave duplicatas. Três operações reprocessam dados e resolvem problemas diferentes:

  • Rewind is for a recoverable pipeline that wrote incorrect data from a known point in time, such as after a bad transformation, malformed input, or a bad code deploy. It restores table data, source offsets, and operator state to a point before the problem and reprocesses only the affected data, keeping operator state. Data before that point is untouched.
  • Full refresh reconstrói uma tabela a partir de todos os dados de origem disponíveis e descarta seu conteúdo atual. Use-o para recalcular tudo do zero ou quando uma alteração de código não for compatível com o estado existente.
  • Reset de ponto de verificação recupera um pipeline cujo ponto de verificação é inválido ou está corrompido, ou que está bloqueado por uma alteração de código incompatível com o ponto de verificação. Ele reseta o ponto de verificação e continua avançando, preservando o conteúdo atual da tabela. O retrocesso não pode recuperar esses casos, porque ele não flexibiliza as regras de compatibilidade do Structured Streaming.

Requisitos​

Obrigatório

Detalhe

Canal

O pipeline deve estar no canal Preview . Consulte Configurar pipelines.

Configuração

Defina a configuração do pipeline pipelines.rewind.betaEnabled como true e, em seguida, execute o pipeline uma vez. Cada fluxo torna-se passível de retrocesso somente após concluir uma atualização com a viagem do tempo ativada.

Modo do pipeline

Trigger and continuous pipelines. O modo de tempo real não é compatível.

Fontes

Tabelas Delta, tabelas de transmissão, Kafka e Auto Loader.

Destinos

Tabelas de transmissão e views materializadas.

Fluxos

Fluxos de transmissão e fluxos de captura de dados de alterações (CDC), incluindo destinos SCD Tipo 1 e SCD Tipo 2. Há suporte para queries com estado, como agregações, joins e desduplicação.

Obrigatório

Detalhe

Canal

O pipeline deve estar no canal Preview . Consulte Configurar pipelines.

Configuração

Defina a configuração do pipeline pipelines.rewind.betaEnabled como true e, em seguida, execute o pipeline uma vez. Cada fluxo torna-se passível de retrocesso somente após concluir uma atualização com a viagem do tempo ativada.

Modo do pipeline

Trigger and continuous pipelines. O modo de tempo real não é compatível.

Fontes

Tabelas Delta, tabelas de transmissão, Kafka e Auto Loader.

Destinos

Tabelas de transmissão e views materializadas.

Fluxos

Fluxos de transmissão e fluxos de captura de dados de alterações (CDC), incluindo destinos SCD Tipo 1 e SCD Tipo 2. Há suporte para queries com estado, como agregações, joins e desduplicação.

Cada fluxo no pipeline deve atender a estes requisitos. Enquanto pipelines.rewind.betaEnabled estiver true, um pipeline que contenha um fluxo que não atenda aos requisitos falha em suas atualizações. Confirme se cada fluxo atende aos requisitos acima antes de ativá-lo.

nota

Um pipeline que tem pipelines.rewind.betaEnabled definido como true não pode retornar para o canal Atual até que o canal Atual seja atualizado para um Runtime que dê suporte ao retrocesso.

Como funcionam a restauração e a repetição​

O retrocesso e a repetição são passos separados.

Rewind restaura cada tabela para a versão que ela continha em um ponto de retrocesso e reinicia os pontos de verificação de transmissão que rastreiam até onde cada fluxo foi lido. As suas transformações não são executadas e nenhum dado de origem é reprocessado.

O reprocessamento ocorre na próxima vez que o pipeline for executado. Ele reprocessa a partir do ponto de restauração usando a definição atual do pipeline, alcança o momento atual e, em seguida, retoma o processamento incremental normal. A restauração não inicia o pipeline; portanto, comece-o você mesmo quando estiver pronto.

O pipeline gera pontos de restauração automaticamente, cerca de uma vez por hora, e os mantém por 7 dias. Um pipeline que você acabou de criar não tem para onde retroceder até que produza o primeiro.

Restaurar um dataset também restaura tudo o que está a jusante dele no mesmo pipeline. A restauração abrange um pipeline. Ele não se coordena com outros pipelines ou com leitores externos das mesmas tabelas, portanto, gerencie-os separadamente.

Rewind a pipeline using the UI​

A interface do usuário e o Genie são as principais formas de usar o retrocesso. A interface do usuário lista os pontos de retrocesso disponíveis e mostra quais datasets cada um afeta antes de você fazer o commit.

  1. Na página do seu pipeline, clique no botão Chevron down icon. ao lado de Run pipeline e clique em Rewind pipeline .
  2. Selecione um ponto de rebobinamento ou use um atalho como Rewind to yesterday ou Rewind to the latest point . Clique em Next .
  3. Select which tables to include. Use the Graph view to select datasets in the pipeline graph, or the List view to select them from a table. Leave Reset all checkpoints selected (the default) to restore source offsets and operator state along with the table data, so the pipeline reprocesses from the rewind point. Clear it to restore the table data only, without reprocessing, for example when you want a table's contents restored but do not want to reprocess the affected data. This setting must be the same for a table and its upstreams, and it cannot be cleared for a flow that reads an external source such as Kafka or Auto Loader. Click Next .
  4. Revise o ponto de retrocesso, a configuração do ponto de verificação e os conjuntos de dados afetados e clique em Rewind .

Inicie o pipeline para reproduzir os dados.

Você pode retroceder repetidamente. Cada retrocesso substitui o anterior, permitindo que você se recupere de uma reprodução com falha retrocedendo para outro lugar.

Retroceda um pipeline usando a API, CLI ou SDK​

Além da UI e do Genie, você pode Trigger um rewind programaticamente com a CLI do Databricks, a REST API ou o Python SDK. Use o caminho programático para automação, como um job de remediação, um suporte de teste ou um orquestrador.

Um rewind é uma única atualização de pipeline cujo corpo da solicitação carrega um rewind_spec. Os campos no rewind_spec definem as mesmas escolhas que você faz na UI: o ponto para o qual retroceder, os datasets para incluir e se deseja fazer Reset nos checkpoints. O replay é a próxima atualização que você envia sem rewind_spec, que é o equivalente de API de iniciar o pipeline. O Rewind não começar o pipeline para você.

Você pode atingir qualquer um dos seguintes:

  • Um ponto de rebobinamento (recomendado). Defina rewind_point_id como um dos pontos de rebobinamento que o pipeline gerou. Essa é a mesma escolha que a interface do usuário e o Genie fazem. O ponto de rebobinamento já registra a versão da tabela e o ponto de verificação a ser restaurado para cada dataset, portanto, o rebobinamento restaura exatamente esse estado. Consulte Encontrar um ponto de rebobinamento.
  • Um Timestamp . Defina rewind_timestamp para um ponto arbitrário no tempo quando nenhum ponto de retrocesso gerado corresponder ao estado desejado. Você escolhe o Timestamp. Consulte Retroceder para um Timestamp.

Encontrar um ponto de restauração​

O pipeline registra cada ponto de restauração como um evento rewind_summary em seu log de eventos. O id do evento é o rewind_point_id. Cada ponto de restauração periódico registra a versão atual de cada tabela no pipeline. Uma execução dry também registra um ponto de restauração, para o estado que ela validou. Para listar os pontos de restauração dos últimos 7 dias, execução uma query como a seguinte:

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

Os pontos de retrocesso periódicos têm a mensagem Periodic rewind summary capturing current table versions. Para desfazer dados ruins, escolha o ponto de retrocesso mais recente cujo timestamp está antes de os dados ruins terem sido escritos.

Se o ID não corresponder a um evento rewind_summary no log de eventos do pipeline, o retrocesso falhará com INVALID_REWIND_SPECIFICATION.REWIND_POINT_NOT_FOUND.

Corpo da solicitação de retrocesso​

Para rebobinar um pipeline, envie uma atualização com um corpo rewind_spec como o seguinte:

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

Rewind com a CLI​

A CLI do Databricks passa o corpo da solicitação pelo --json. Para retroceder um pipeline, execução o seguinte:

O comando databricks pipelines start-update descarta rewind_point_id do corpo da solicitação. Envie a solicitação com databricks api post em vez disso, que passa o corpo sem alterações:

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

Para repetir os dados, comece uma atualização simples sem rewind_spec:

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

Para verificar uma atualização até que ela atinja um estado terminal, execute o seguinte:

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

Retroceder com a API REST​

Para retroceder um pipeline, envie uma solicitação POST para o endpoint começar uma atualização com o corpo rewind_spec. Por exemplo:

Bash
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
    }
  }'

Retroceder com o Python SDK​

O auxiliar RewindSpec tipado não inclui rewind_point_id. Envie a solicitação com o cliente REST autenticado do SDK em vez disso. Isso funciona em todas as versões do SDK, incluindo o databricks-sdk empacotado com o Databricks Runtime e o compute serverless. Para restaurar um pipeline, faça a execução do seguinte:

Python
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"])

Referência de especificação de retrocesso​

A tabela a seguir descreve os campos no corpo de rewind_spec.

campo

Descrição

rewind_point_id

O id de um evento rewind_summary para restaurar. Consulte Localizar um ponto de restauração. Defina exatamente um entre rewind_point_id ou rewind_timestamp.

rewind_timestamp

O ponto no tempo para restaurar, no formato UTC yyyy-MM-dd HH:mm:ss. Consulte Retroceder para um timestamp. Defina exatamente um de rewind_point_id ou rewind_timestamp.

datasets

A lista de datasets a serem revertidos. Omita a lista para retroceder todos os datasets no pipeline, cada um com cascade e reset_checkpoints definidos como true.

datasets[].identifier

O nome totalmente qualificado do dataset a ser retrocedido, no formato catalog.schema.table.

datasets[].cascade

Se deseja também retroceder cada dataset a jusante deste no pipeline. Especifique este campo para cada dataset. Para um pipeline linear, defina como true no dataset raiz para retroceder todo o pipeline.

datasets[].reset_checkpoints

Se deseja Reset os deslocamentos da fonte de transmissão e o estado do operador para que a repetição processe novamente a partir do ponto de rebobinamento. Esta é a configuração Reset todos os checkpoints da UI. default para true. Defina como false para restaurar apenas os dados da tabela, sem reprocessar.

dry_run

Se validar o rewind sem alterar nenhuma tabela. É default false. Consulte Praticar um rewind com uma dry execução.

campo

Descrição

rewind_point_id

O id de um evento rewind_summary para restaurar. Consulte Localizar um ponto de restauração. Defina exatamente um entre rewind_point_id ou rewind_timestamp.

rewind_timestamp

O ponto no tempo para restaurar, no formato UTC yyyy-MM-dd HH:mm:ss. Consulte Retroceder para um timestamp. Defina exatamente um de rewind_point_id ou rewind_timestamp.

datasets

A lista de datasets a serem revertidos. Omita a lista para retroceder todos os datasets no pipeline, cada um com cascade e reset_checkpoints definidos como true.

datasets[].identifier

O nome totalmente qualificado do dataset a ser retrocedido, no formato catalog.schema.table.

datasets[].cascade

Se deseja também retroceder cada dataset a jusante deste no pipeline. Especifique este campo para cada dataset. Para um pipeline linear, defina como true no dataset raiz para retroceder todo o pipeline.

datasets[].reset_checkpoints

Se deseja Reset os deslocamentos da fonte de transmissão e o estado do operador para que a repetição processe novamente a partir do ponto de rebobinamento. Esta é a configuração Reset todos os checkpoints da UI. default para true. Defina como false para restaurar apenas os dados da tabela, sem reprocessar.

dry_run

Se validar o rewind sem alterar nenhuma tabela. É default false. Consulte Praticar um rewind com uma dry execução.

atenção

Quando um dataset raiz lê uma fonte externa como Kafka ou Auto Loader, mantenha reset_checkpoints definido como true. A interface do usuário impede a limpeza de Reset todos os pontos de verificação para essas fontes, mas a API aceita uma solicitação que define reset_checkpoints como false. A solicitação é aceita e o retrocesso pode começar, mas o pipeline falha depois. Limite reset_checkpoints: false a datasets que um fluxo de transmissão não lê.

Restaurar para um timestamp​

Use rewind_timestamp quando nenhum ponto de retrocesso gerado corresponder ao estado desejado, por exemplo, para retroceder para um ponto entre dois pontos de retrocesso. Você mesmo fornece o timestamp. Siga duas regras:

  • Escolha um ponto estritamente entre dois commits, não um que coincida com um commit. Um timestamp que coincide com um commit pode fazer a reprodução falhar. Use o ponto médio entre o último commit bom e o primeiro commit ruim.
  • Em um pipeline que não foi restaurado ou atualizado, verifique o feed de dados de alteração da versão 1. A versão 0 não registra dados de alteração. Se a versão 1 for mais antiga que o período de retenção de arquivo da tabela (7 dias default), a verificação falha, portanto, comece da versão mais antiga confirmada dentro desse período.

A query abaixo, que encontra os tempos de commit, lê o feed de dados de alteração (CDF) da tabela de pipeline para a qual você deseja retroceder, portanto, o CDF deve estar habilitado nessa tabela. O retrocesso em si não requer CDF. Os LakeFlow Pipelines habilitam o CDF em suas tabelas por default, então a query funciona sem configuração adicional. Se o CDF tiver sido desabilitado na tabela, use um ponto de retrocesso ou DESCRIBE HISTORY para escolher um ponto.

Corresponda o filtro _change_type à falha que você está investigando: para inserções ou atualizações incorretas, mantenha 'insert' e 'update_postimage'; para uma exclusão errônea, filtre em 'delete', porque table_changes mostra as linhas excluídas apenas sob esse tipo de alteração. Para compute um Timestamp de ponto médio a partir de um predicado que seleciona os dados incorretos, execução uma query como a seguinte:

SQL
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 o pipeline já foi rebobinado ou refresh, comece a varredura a partir do último RESTORE ou commit de refresh para que ele não ultrapasse a operação anterior.

Ensaie um retrocesso com uma execução de teste​

Para validar um retrocesso sem alterar nenhuma tabela, defina dry_run como true. A execução é feita através da validação para um estado de terminal COMPLETED sem modificar os dados. Use uma execução de teste para confirmar se o pipeline aceita o retrocesso e, em seguida, envie a solicitação novamente com dry_run definido como false.

Uma execução de teste registra o rebobinamento que validou como um evento rewind_summary. Para aplicar exatamente esse rebobinamento, por exemplo, após uma execução de teste com rewind_timestamp, envie o rebobinamento real com o id do evento como rewind_point_id. Uma execução de teste que falha não bloqueia atualizações posteriores.

Ao executar ciclos sucessivos de retrocesso e repetição na automação, execute um refresh completo entre eles. Empilhar um segundo retrocesso e repetição em um pipeline sem um refresh completo entre eles pode duplicar as linhas reprocessadas em tabelas de transmissão downstream.

After a rewind​

Uma falha de repetição deixa o pipeline retrocedido, mas parado. Corrija o código ou os dados de origem e comece o pipeline para tentar novamente, ou retroceda para um ponto diferente. O pipeline não sofre reversão por conta própria, e os erros aparecem por meio dos diagnósticos padrão do pipeline e do log de eventos.

O replay só é bem-sucedido se a definição atual do pipeline for compatível com o estado restaurado; o rewind não flexibiliza as regras de compatibilidade do Structured Streaming. Para saber quais alterações são compatíveis, consulte Tipos de alterações em consultas do Structured Streaming. As visualizações materializadas seguem a semântica de lotes e toleram alterações de esquema mais amplas, mas ainda falham se uma dependência for incompatível.

Um retrocesso que falha no meio do processo pode deixar o pipeline parcialmente retrocedido. Você tem duas opções:

  1. Volte novamente para o mesmo ponto ou para um diferente, e o pipeline convergerá para esse ponto.
  2. Para forçar o pipeline a começar uma atualização normal apesar do retrocesso incompleto, defina pipelines.allowUpdateAfterIncompleteRewind como true e reinicie o pipeline.
nota

Se um rewind falhar ou se você o enviar com um ponto de rewind inválido, o pipeline bloqueará todas as atualizações subsequentes que não sejam de rewind, incluindo um full refresh, até que você emita novamente um rewind válido. Ao Trigger rewinds programaticamente, confirme se rewind_point_id ou rewind_timestamp se refere a um ponto válido antes de enviar a solicitação.

Até onde é possível retroceder​

Os pontos de restauração são mantidos por 7 dias. Dentro dessa janela, uma restauração falha se os dados necessários já tiverem sido removidos. Verifique estes itens antes de depender da restauração:

  • VACUUM ou uma delta.deletedFileRetentionDuration curta sobre suas tabelas. Consulte Trabalhar com a história da tabela.
  • Retenção de origem menor do que a janela pela qual você deseja retroceder, como um tópico do Kafka que mantém um dia.

Pipelines com várias origens ou longas cadeias de dependência precisam de mais retenção, porque cada tabela e ponto de verificação precisa alcançar um ponto consistente.

Limitações​

  • Não é possível restaurar um pipeline para um ponto anterior a um refresh completo.
  • Os pontos de retrocesso são mantidos por 7 dias, e um retrocesso falha se o histórico da tabela ou os dados de origem necessários para alcançar o ponto já tiverem sido removidos, por exemplo, por VACUUM ou retenção curta da origem. Consulte Até onde você pode retroceder.
  • O modo em tempo real não é compatível.
  • O Kinesis, o Pulsar, o Google Pub/Sub e as fontes de dados personalizadas criadas com as APIs de fonte de dados do DSv2 ou do Python não são compatíveis como fontes.
  • Sinks externos e sinks personalizados não são compatíveis, incluindo sinks definidos com create_sink(). Consulte Usar sinks em pipelines.
  • As tabelas de transmissão que usam um filtro de linha ou uma máscara de coluna não podem ser retrocedidas. Consulte Aplicar manualmente filtros de linha e máscaras de coluna.
  • Stateful rewind requires the RocksDB armazenamento do estado, which pipelines use by default. A rewind fails for a flow configured with a different armazenamento do estado.
  • Algumas tabelas de transmissão AUTO CDC precisam de um refresh antes de poderem ser rebobinadas. O pipeline notifica você quando você solicita o retrocesso.
  • Materialized views might fully recompute rather than incrementally refresh after a rewind. See Incremental refresh for materialized views.
  • O recurso de retrocesso (rewind) abrange um único pipeline e não se coordena com leitores externos ou outros pipelines que leiam as mesmas tabelas.
  • O recurso de restauração não restaura o código do pipeline, a configuração do pipeline ou os metadados de objetos do Unity Catalog, como tags e concessões.

Outros recursos​

Nesta página