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

Fluxos REPLACE WHERE para tabelas de transmissão autônomas

Os fluxos REPLACE WHERE recalculam e substituem um subconjunto alvo de uma tabela de transmissão autônoma sem processar novamente toda a história da tabela. Eles lidam com dados que chegam atrasados, reprocessamento upstream, evolução do esquema e preenchimentos retroativos.

Com um fluxo REPLACE WHERE, você define um predicado na tabela de destino. Todas as linhas que correspondem ao predicado são excluídas e substituídas pela reavaliação da query de origem para esse mesmo intervalo de predicado. As linhas que não correspondem ao predicado não são alteradas.

Requisitos​

Os fluxos REPLACE WHERE têm os seguintes requisitos:

  • O Databricks recomenda o Unity Catalog e compute serverless. Atualização refresh é suportada apenas em compute Serverless.

Quando usar fluxos REPLACE WHERE​

Use o REPLACE WHERE fluxos para os seguintes cenários:

  • Processamento em lotes incremental sem semântica de transmissão : Processe novas linhas em lotes sem gerenciar conceitos de transmissão, como marcas d'água.
  • Processamento seletivo : Recalcula apenas as linhas que correspondem a um predicado, deixando todas as outras linhas intocadas.
  • Cenários além das capacidades padrão de view materializada :
    • Tabelas de destino com retenção mais longa que a origem
    • Como evitar o recalculo quando uma tabela de dimensões é alterada
    • Evolução do esquema sem recalcular toda a história

Criar um fluxo REPLACE WHERE​

Use a cláusula FLOW REPLACE WHERE em linha com CREATE OR REFRESH STREAMING TABLE:

SQL
CREATE OR REFRESH STREAMING TABLE orders_enriched
SCHEDULE EVERY 1 DAY
FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
SELECT
  o.order_id,
  o.date,
  o.region,
  p.product_name,
  o.qty,
  o.price
FROM orders_fct o
JOIN product_dim p
  ON o.product_id = p.product_id;

Durante o refresh, todas as linhas na tabela de destino que correspondem ao predicado são excluídas, a query de origem é recalculada para esse mesmo intervalo de predicado e os novos resultados são inseridos. Neste exemplo, todas as linhas dos últimos 7 dias são excluídas de orders_enriched e recalculadas usando a query de origem.

Como alternativa, use a sintaxe de formato longo CREATE FLOW para definir a tabela de transmissão e o fluxo REPLACE WHERE em instruções separadas:

SQL
CREATE STREAMING TABLE orders_enriched;

CREATE FLOW orders_enriched AS
INSERT INTO orders_enriched BY NAME
REPLACE WHERE date >= date_add(current_date(), -7)
SELECT
  o.order_id,
  o.date,
  o.region,
  p.product_name,
  o.qty,
  o.price
FROM orders_fct o
JOIN product_dim p
  ON o.product_id = p.product_id;

Ambas as formas são equivalentes. Para a referência CREATE FLOW completa, consulte CREATE FLOW (pipelines).

Você não precisa adicionar o predicado à query de origem. O engine do pipeline o aplica automaticamente ao ler da fonte.

nota

BY NAME é obrigatório. Ele garante que as colunas sejam combinadas pelo nome em vez da posição.

Preenchimento de data histórica​

Para gravar linhas históricas ou corrigidas na tabela de destino fora dos refresh agendados, escolha entre dois mecanismos com base onde a data histórica reside:

  • Substituições de predicado : Faça a execução novamente da query de origem do fluxo para um intervalo de predicado único usando REFRESH STREAMING TABLE ... WHERE. Use quando a data histórica vier da mesma fonte que a data incremental.
  • Declarações DML : Insira diretamente na tabela de destino, ignorando o fluxo. Use quando a data histórica estiver em uma fonte diferente da data incremental.

Substituições de predicados​

Substitua o predicado REPLACE WHERE do fluxo para um refresh único sem modificar a definição da tabela. As substituições de predicado são únicas, aplicam-se apenas a esse refresh e não afetam os refresh agendados subsequentes.

Use a cláusula WHERE com REFRESH STREAMING TABLE:

SQL
REFRESH STREAMING TABLE orders_enriched WHERE date BETWEEN '2020-01-01' AND '2024-12-31';

O fluxo exclui apenas as linhas que correspondem ao predicado de substituição e as recomputa a partir da origem. O predicado estático do fluxo permanece inalterado; portanto, o próximo refresh agendado usa o predicado original.

Você pode combinar a substituição com SYNC ou ASYNC. Por exemplo, para começar o refresh em segundo plano e retornar imediatamente com um link para a atualização do pipeline:

SQL
REFRESH STREAMING TABLE orders_enriched WHERE date >= date_add(current_date(), -30) ASYNC;
nota

A cláusula WHERE é compatível apenas em uma tabela de transmissão criada com uma cláusula FLOW REPLACE WHERE . Especificá-la em qualquer outra tabela de transmissão retorna um erro. Consulte REFRESH (MATERIALIZED VIEW ou STREAMING TABLE).

Instruções DML​

Execução as instruções DML diretamente na tabela de destino para carregar as linhas provenientes de uma fonte diferente do fluxo:

SQL
INSERT INTO orders_enriched
SELECT *
FROM orders_enriched_legacy
WHERE date < '2025-01-01';

Comportamento de refresh completo​

Um refresh completo de um fluxo REPLACE WHERE reexecuta a query de origem usando apenas o predicado definido pelo fluxo. Ela exclui permanentemente todas as linhas no destino que não correspondem a esse predicado, incluindo linhas que esse predicado substitui ou instruções DML inseridas anteriormente fora do intervalo de predicado definido.

atenção

Um refresh completo limpa todos os dados existentes e reexecuta o fluxo usando apenas o predicado definido. Se um pipeline estiver em execução há um ano com um predicado de 7 dias, um refresh completo fará com que a tabela contenha apenas os últimos 7 dias de dados. Todas as linhas mais antigas são excluídas permanentemente.

SQL
REFRESH STREAMING TABLE orders_enriched FULL;

Para evitar refresh completos em uma tabela, defina a propriedade da tabela pipelines.reset.allowed como false:

SQL
CREATE OR REFRESH STREAMING TABLE orders_enriched
  TBLPROPERTIES (pipelines.reset.allowed = 'false')
  FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
  ...

Refresh incremental​

Os fluxos REPLACE WHERE usam refresh incremental sempre que possível, reprocessando apenas os dados de origem que foram alterados desde o último refresh, em vez de recalcular toda a janela de substituição. O refresh incremental requer compute serverless.

Quando o refresh incremental se aplica​

Tudo o seguinte deve ser verdadeiro:

  • O pipeline é executado em compute serverless.
  • O formato da query é compatível. Veja Refresh incremental para o conjunto de operadores compatíveis.
  • O predicado faz referência às colunas base de uma tabela de origem. Os predicados em valores derivados, como saídas de função agregada ou de janela, não podem ser enviados para uma origem, o que desativa o refresh incremental.
  • Nenhum DML externo modificou linhas na janela de substituição atual. O DML que modifica linhas fora da janela atual não é afetado.
  • A janela de substituição atual não inclui linhas que o predicado anterior excluiu. Se você ampliar o predicado para cobrir um intervalo que não foi processado anteriormente, esse refresh retornará ao recálculo completo. Os refresh subsequentes são elegíveis para refresh incremental novamente.
  • O predicado é determinístico. Predicados usando funções não determinísticas como rand() desabilitam o refresh incremental. Funções temporais como current_date() são permitidas.

O primeiro refresh de qualquer fluxo é sempre um cálculo completo. Se alguma condição não for atendida, esse refresh voltará ao recompute completo da janela de substituição atual.

Práticas recomendadas para refresh incremental​

Siga estas diretrizes para que os fluxos REPLACE WHERE permaneçam elegíveis para refresh incremental.

Use um limite inferior móvel​

Os predicados com um limite inferior móvel permanecem qualificados para refresh incremental indefinidamente.

SQL
FLOW REPLACE WHERE date >= date_add(current_date(), -7)

Um limite superior móvel, como date BETWEEN date_add(current_date(), -7) AND current_date(), pode deslocar a janela para incluir linhas previamente excluídas, disparando um fallback único para recomputação completa.

Incluir a coluna de predicado no GROUP BY​

Ao agregar, inclua a coluna de predicado em GROUP BY para que o mecanismo possa empurrar o predicado para abaixo da agregação.

SQL
FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
SELECT date, region, SUM(amount) AS total
FROM sales
GROUP BY date, region;

Se a coluna de predicado estiver faltando em GROUP BY, o predicado não poderá ser enviado abaixo da agregação e a fonte será verificada na íntegra.

Incluir a coluna de predicado nas chaves de join​

Inclua a coluna predicada na condição de join para que o motor possa podar todas as fontes unidas.

SQL
FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
SELECT f.date, f.user_id, d.region, f.revenue
FROM fact f
JOIN dim d ON f.date = d.date AND f.user_id = d.user_id;

Se uma tabela unida não exibir a coluna de predicado, essa tabela será escaneada completamente a cada refresh.

Diagnosticar fallback para recompute completo​

Quando um refresh retrocede para o recálculo completo, o motivo é relatado no evento planning_information para o fluxo. Consulte Objeto IssueType no esquema de log de eventos do pipeline. A tabela a seguir lista os motivos relatados no evento:

Motivo

Significado

EXTERNAL_CHANGE_IN_REPLACE_WINDOW

Um DML externo modificou linhas na janela de substituição atual.

REPLACE_WHERE_NOT_DETERMINISTIC

O predicado usa expressões não determinísticas.

PRIOR_REPLACE_WHERE_NOT_DETERMINISTIC

O refresh anterior usou um predicado não determinístico.

PRIOR_REPLACE_WHERE_PREDICATE_MISSING

Não foi possível recuperar o predicado usado na atualização anterior. Isso pode ocorrer se o predicado não foi serializado corretamente durante essa atualização.

STALE_TARGET_DATA_IN_WINDOW

O predicado foi alterado entre as atualizações e a janela atual inclui linhas que o predicado anterior excluiu.

CANNOT_PUSH_PREDICATE

O predicado não pode ser enviado para pelo menos uma fonte.

REPLACE_WHERE_BACKFILL

A atualização é um preenchimento retroativo que usa uma substituição de predicado única para preencher data histórica ou corrigir gravações anteriores.

Motivo

Significado

EXTERNAL_CHANGE_IN_REPLACE_WINDOW

Um DML externo modificou linhas na janela de substituição atual.

REPLACE_WHERE_NOT_DETERMINISTIC

O predicado usa expressões não determinísticas.

PRIOR_REPLACE_WHERE_NOT_DETERMINISTIC

O refresh anterior usou um predicado não determinístico.

PRIOR_REPLACE_WHERE_PREDICATE_MISSING

Não foi possível recuperar o predicado usado na atualização anterior. Isso pode ocorrer se o predicado não foi serializado corretamente durante essa atualização.

STALE_TARGET_DATA_IN_WINDOW

O predicado foi alterado entre as atualizações e a janela atual inclui linhas que o predicado anterior excluiu.

CANNOT_PUSH_PREDICATE

O predicado não pode ser enviado para pelo menos uma fonte.

REPLACE_WHERE_BACKFILL

A atualização é um preenchimento retroativo que usa uma substituição de predicado única para preencher data histórica ou corrigir gravações anteriores.

Exemplos​

Os exemplos a seguir mostram padrões comuns de fluxo REPLACE WHERE.

Exemplo 1: manter agregados históricos de uma origem com retenção limitada​

Este exemplo mantém agregados diários indefinidamente, mesmo depois que os dados brutos saem da tabela de origem (retenção de 3 dias):

SQL
CREATE OR REFRESH STREAMING TABLE events_agg
FLOW REPLACE WHERE date >= date_add(current_date(), -3) BY NAME
SELECT
  date,
  key,
  SUM(val) AS agg
FROM events_raw
GROUP BY ALL;

Exemplo 2: Impedir o recálculo quando uma tabela de dimensões é alterada​

Este exemplo mantém as linhas de fatos históricos inalteradas quando os atributos de dimensão mudam:

SQL
CREATE OR REFRESH STREAMING TABLE fact_dim_join
FLOW REPLACE WHERE date >= date_add(current_date(), -1) BY NAME
SELECT
  f.date,
  f.user_id,
  d.region,
  f.revenue
FROM fact_table f
JOIN dim_users d
  ON f.user_id = d.user_id;

Se a região de um usuário mudar, apenas as linhas recentes serão recalculadas. As linhas históricas mantêm o valor da região no momento em que foram gravadas.

Exemplo 3: Adicionar uma nova métrica sem recalcular a história completa​

Este exemplo mostra como evoluir uma definição de tabela e preencher retroativamente apenas um intervalo direcionado:

  1. Defina a tabela inicial:

    SQL
    CREATE OR REFRESH STREAMING TABLE clickstream_daily
    FLOW REPLACE WHERE event_date >= date_add(current_date(), -7) BY NAME
    SELECT
      event_date,
      page_id,
      COUNT(*) AS clicks
    FROM clickstream_raw
    GROUP BY ALL;

  2. Atualize a query para adicionar uniq_users:

    SQL
    CREATE OR REFRESH STREAMING TABLE clickstream_daily
    FLOW REPLACE WHERE event_date >= date_add(current_date(), -7) BY NAME
    SELECT
      event_date,
      page_id,
      COUNT(*) AS clicks,
      COUNT(DISTINCT user_id) AS uniq_users
    FROM clickstream_raw
    GROUP BY ALL;

    As linhas anteriores à janela de 7 dias contêm NULL para uniq_users.

Exemplo 4: Iterar em uma pequena janela antes de preencher a história completa​

Este exemplo mostra como validar a lógica de query em uma pequena janela de dados antes de processar todo o intervalo histórico.

Comece com uma janela curta para validar métricas e iterar sobre a lógica de negócios com custos de compute mais baixos:

SQL
CREATE OR REFRESH STREAMING TABLE revenue_attribution
FLOW REPLACE WHERE event_date >= date_add(current_date(), -7) BY NAME
SELECT
  event_date,
  campaign_id,
  SUM(revenue) AS total_revenue
FROM marketing_events
GROUP BY ALL;

Uma janela curta recalcula apenas os últimos 7 dias em cada refresh, então revise a query tantas vezes quantas forem necessárias antes de se commit com uma execução histórica completa.

Depois que a query estiver finalizada, use DML para preencher o intervalo histórico completo:

SQL
INSERT INTO revenue_attribution
SELECT
  event_date,
  campaign_id,
  SUM(revenue) AS total_revenue
FROM marketing_events
WHERE event_date < date_add(current_date(), -7)
GROUP BY ALL;
Nesta página