Nahradit toky WHERE pro samostatně streamované tabulky

Toky REPLACE WHERE znovu vypočítají a přepíšou vybranou podmnožinu samostatné tabulky datového proudu, aniž by bylo nutné znovu zpracovat celou historii tabulky. Zpracovávají opožděně doručená data, opakované zpracování upstreamových dat, evoluci schématu a zpětné doplnění dat.

S tokem REPLACE WHERE definujete predikát v cílové tabulce. Všechny řádky odpovídající predikátu se odstraní a nahradí opětovným vyhodnocením zdrojového dotazu pro stejnou oblast predikátu. Řádky, které neodpovídají predikátu, zůstanou nedotčené.

Requirements

Toky REPLACE WHERE mají následující požadavky:

• Databricks doporučuje Katalog Unity a bezserverové výpočetní prostředky. Přírůstková aktualizace se podporuje jenom na výpočetních prostředcích bez serveru.

Kdy použít toky typu REPLACE WHERE

Toky REPLACE WHERE použijte pro následující scénáře:

• Přírůstkové dávkové zpracování bez sémantiky streamování: Zpracování nových řádků v dávkách bez správy konceptů streamování, jako jsou vodoznaky.
• Selektivní opětovné zpracování: Znovu zkompilujte pouze řádky, které odpovídají predikátu, a přitom ponechejte všechny ostatní řádky nedotčené.
• Scénáře nad rámec standardních materializovaných možností zobrazení:
  • Cílové tabulky s delší dobou uchovávání než zdroj
  • Zabránění přepočítání při změně tabulky dimenzí
  • Vývoj schématu bez opětovného dokončování celé historie

Vytvoření toku REPLACE WHERE

Použijte klauzuli FLOW REPLACE WHERE přímo v řádku s CREATE OR REFRESH STREAMING TABLE:

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;

Během aktualizace se odstraní všechny řádky v cílové tabulce, které odpovídají predikátu, zdrojový dotaz se přepočítá pro stejnou oblast predikátu a vloží se nové výsledky. V tomto příkladu se všechny řádky z posledních 7 dnů odstraní orders_enriched a přepočítají pomocí zdrojového dotazu.

Do zdrojového dotazu nemusíte přidávat predikát. Modul pipeline jej při čtení ze zdroje automaticky použije.

Note

BY NAME je povinné. Zajišťuje, že sloupce se budou shodovat podle názvu, nikoli podle pozice.

Obnovení historických dat

Pokud chcete provést obnovení, spusťte příkazy DML přímo v cílové tabulce:

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

Chování úplné aktualizace

Úplná aktualizace toku REPLACE WHERE znovu spustí zdrojový dotaz pouze pomocí aktuálního predikátu. Řádky vložené příkazy DML mimo aktuální rozsah predikátu jsou trvale odstraněny.

Výstraha

Úplná aktualizace vymaže všechna existující data a znovu spustí tok pouze pomocí definovaného predikátu. Pokud pipeline běží už rok se sedmidenním predikátem, úplná aktualizace způsobí, že tabulka obsahuje pouze data za posledních 7 dní. Všechny starší řádky se trvale odstraní.

REFRESH STREAMING TABLE orders_enriched FULL;

Pokud chcete zabránit úplným aktualizacím v tabulce, nastavte vlastnost pipelines.reset.allowed tabulky na false:

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

Přírůstková aktualizace

Toky REPLACE WHERE používají přírůstkovou aktualizaci, pokud je to možné, znovu zpracovávají pouze zdrojová data, která se od poslední aktualizace změnila, a nekomputují celé okno nahrazení. Přírůstková aktualizace vyžaduje výpočetní prostředky bez serveru.

Kdy se použije přírůstková aktualizace

Musí být splněny všechny následující podmínky:

• Kanál běží na výpočetních prostředcích bez serveru.
• Obrazec dotazu je podporovaný. Viz Přírůstková aktualizace, kde najdete informace o podporované sadě operátorů.
• Predikát odkazuje na základní sloupce ze zdrojové tabulky. Predikáty na odvozených hodnotách, jako jsou výstupy agregace nebo funkce okna, nelze odeslat do zdroje, což zakáže přírůstkovou aktualizaci.
• V aktuálním okně nahrazení nebyly změněny žádné externí řádky DML. Objekt DML, který upravuje řádky mimo aktuální okno, není ovlivněn.
• Aktuální okno nahrazení neobsahuje řádky, které předchozí predikát vyloučil. Pokud rozšíříte predikát tak, aby pokrývala rozsah, který nebyl dříve zpracován, vrátí se tato aktualizace zpět k úplnému výpočtu. Další aktualizace lze znovu provádět jako přírůstkové.
• Predikát je deterministický. Predikáty používající nedeterministické funkce, jako je rand(), zakazují přírůstkovou aktualizaci. Dočasné funkce, jako current_date() jsou povolené.

První aktualizace jakéhokoli toku je vždy úplný výpočet. Pokud některá podmínka není splněná, vrátí se tato aktualizace zpět do úplného výpočtu aktuálního okna nahrazení.

Osvědčené postupy pro přírůstkovou aktualizaci

Postupujte podle těchto pokynů, aby toky REPLACE WHERE zůstaly způsobilé pro přírůstkovou aktualizaci.

Použijte pohyblivou dolní mez

Predikáty s pohyblivým dolním mezem zůstávají způsobilé pro přírůstkovou aktualizaci na neomezenou dobu.

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

Posouvající se horní mez, například date BETWEEN date_add(current_date(), -7) AND current_date(), může posunout okno tak, aby zahrnovalo dříve vyloučené řádky, což vyvolá jednorázový návrat k úplnému přepočtu.

Zahrňte sloupec predikátů do GROUP BY

Při agregování zahrňte do sloupec predikátu, aby stroj mohl přesunout predikát pod agregaci.

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

Pokud ve GROUP BY chybí sloupec predikátu, predikát nelze přesunout pod agregaci a zdroj se prochází celý.

Zahrnout sloupec s predikátem do klíčů spojení

Do podmínky spojení zahrňte sloupec predikátu, aby modul mohl odfiltrovat všechny připojené zdroje.

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;

Pokud propojená tabulka nezpřístupňuje sloupec predikátu, je tato tabulka při každé aktualizaci prohledána v plném rozsahu.

Diagnostika přechodu na úplný přepočet

Když aktualizace přejde na úplný přepočet, důvod je uveden v události planning_information pro tok. Viz Sledování protokolů událostí kanálu. Následující tabulka uvádí důvody nahlášené v události:

Reason	Význam
EXTERNAL_CHANGE_IN_REPLACE_WINDOW	Externí příkaz DML upravil řádky v aktuálním okně pro nahrazování.
REPLACE_WHERE_NOT_DETERMINISTIC	Predikát používá ne deterministické výrazy.
PRIOR_REPLACE_WHERE_NOT_DETERMINISTIC	Předchozí aktualizace používala ne deterministický predikát.
UNSUPPORTED_REPLACE_WHERE_PREDICATE	Predikát nelze použít na žádném zdroji, aktuální okno obsahuje řádky, které nebyly zpracovány předchozím predikátem, nebo běh používá přepsání predikátu.

Příklady

Následující příklady ukazují běžné vzory toku REPLACE WHERE .

Příklad 1: Zachování historických agregací ze zdroje omezeného uchovávání

Tento příklad uchovává denní agregace po neomezeně dlouhou dobu, a to i poté, co jsou nezpracovaná data ze zdrojové tabulky odstraněna po uplynutí 3denní doby uchovávání:

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;

Příklad 2: Zabránění přepočítaní, když se změní tabulka dimenzí

Tento příklad uchovává historické řádky faktů beze změny při změně atributů dimenzí:

CREATE OR REFRESH STREAMING TABLE fact_dim_join
FLOW REPLACE WHERE f.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;

Pokud se oblast uživatele změní, přepočítají se jenom poslední řádky. Historické řádky uchovávají hodnotu oblasti v době, kdy byly zapsány.

Příklad 3: Přidání nové metriky bez opětovného dokončování úplné historie

Tento příklad ukazuje, jak upravovat definici tabulky a zpětně doplnit data pouze pro vybraný rozsah:

1. Definujte počáteční tabulku:

   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. Aktualizujte dotaz tak, aby přidal uniq_users:

   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;
   

   Řádky starší než sedmidenní období obsahují hodnotu NULL pro uniq_users.

Příklad 4: Iterujte nad malým časovým oknem před zpětným doplněním celé historie

Tento příklad ukazuje, jak před zpracováním úplného historického rozsahu ověřit logiku dotazu v malém okně dat.

Začněte krátkým oknem pro ověření metrik a iterací obchodní logiky s nižšími náklady na výpočetní prostředky:

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;

Krátké časové okno při každém obnovení přepočítá pouze posledních 7 dní, proto dotaz před spuštěním úplného historického běhu upravujte tolikrát, kolikrát je potřeba.

Po dokončení dotazu použijte DML k obnovení úplného historického rozsahu:

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;