REPLACE WHERE folyamatok önálló streamelési táblákhoz

A REPLACE WHERE folyamok újraszámítják és felülírják egy önálló streamelési tábla célzott részhalmazát anélkül, hogy a tábla teljes előzményét újra kellene feldolgozni. Kezelik a későn érkező adatokat, a felsőbb rétegbeli újrafeldolgozást, a sémafejlődést és a háttérbetöltéseket.

A REPLACE WHERE folyamattal meghatároz egy predikátumot a céltáblán. A predikátumnak megfelelő összes sort törlik, majd az ugyanarra a predikátumtartományra vonatkozó forráslekérdezés újbóli kiértékelésének eredményével helyettesítik. A predikátumnak nem megfelelő sorok érintetlenek maradnak.

Requirements

A REPLACE-folyamatokra WHERE a következő követelmények vonatkoznak:

• A Databricks a Unity Catalog és a kiszolgáló nélküli számítás használatát javasolja. A növekményes frissítés csak kiszolgáló nélküli számítás esetén támogatott.

Mikor érdemes a REPLACE WHERE folyamatokat használni?

A REPLACE WHERE folyamatokat a következő forgatókönyvekhez használhatja:

• Növekményes kötegelt feldolgozás streaming szemantika nélkül: Új sorok feldolgozása kötegekben, streamingfogalmak — például vízjelek — kezelése nélkül.
• Szelektív újrafeldolgozás: Csak a predikátumnak megfelelő sorok újrafordítása, miközben az összes többi sort érintetlenül hagyja.
• A standard materializált nézet képességein túli forgatókönyvek:
  • A forrásénál hosszabb megőrzési idejű céltáblák
  • A dimenziótáblák módosításakor történő újraszámítás megakadályozása
  • Sémafejlődés a teljes előzmények újrafordítása nélkül

REPLACE WHERE folyamat létrehozása

Használja a(z) FLOW REPLACE WHERE záradékot egy sorban a(z) CREATE OR REFRESH STREAMING TABLE elemmel:

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;

A frissítés során a céltábla minden olyan sora törlődik, amely megfelel a predikátumnak, a forrás lekérdezés ugyanazon predikátumtartományhoz lesz újrafordítve, és az új eredmények be lesznek szúrva. Ebben a példában az elmúlt 7 nap összes sorát törlik a(z) orders_enriched elemből, majd a forráslekérdezés alapján újraszámítják.

Alternatívaként a hosszú formátumú CREATE FLOW szintaxis segítségével definiáljuk a streaming táblát és a REPLACE WHERE folyamatot külön állításokban:

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;

Mindkét űrlap egyenértékű. A teljes CREATE FLOW hivatkozásért lásd: CREATE FLOW (pipelines).

Nem kell hozzáadnia a predikátumot a forrás lekérdezéshez. A pipeline-motor automatikusan alkalmazza azt a forrás olvasásakor.

Megjegyzés:

A BY NAME használata kötelező. Biztosítja, hogy az oszlopok névvel és nem pozícióval egyeznek.

Előzményadatok visszatöltése

Ha az ütemezett frissítéseken kívül szeretne előzménysorokat vagy javított sorokat írni a céltáblába, válasszon két mechanizmust az előzményadatok helyétől függően:

• Predikátum felülbírálása: Futtassa újra a folyamat forráslekérdezését egy egyszeri predikátumtartományhoz a REFRESH STREAMING TABLE ... WHERE használatával. Akkor használható, ha az előzményadatok ugyanabból a forrásból származnak, mint a növekményes adatok.
• DML-utasítások: Szúrja be közvetlenül a céltáblába, megkerülve a folyamatot. Akkor használható, ha az előzményadatok más forrásban élnek, mint a növekményes adatok.

Predikátum felülírások

A folyamat REPLACE WHERE predikátumának felülbírálása egyetlen frissítés erejéig, a tábladefiníció módosítása nélkül. A predikátum-felülbírálások egyszeriek, csak az adott frissítésre vonatkoznak, és nem érintik a későbbi ütemezett frissítéseket.

Használja a WHERE záradékot a REFRESH STREAMING TABLE elemmel:

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

A folyamat csak az override predikátumnak megfelelő sorokat törli, és azokat a forrásból újraszámítja. A folyamat statikus predikátuma változatlan, ezért a következő ütemezett frissítés az eredeti predikátumot használja.

A felülírást kombinálhatja a(z) SYNC vagy ASYNC elemmel. Ha például a háttérben szeretné elindítani a frissítést, és azonnal visszatér a folyamatfrissítésre mutató hivatkozással:

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

Megjegyzés:

A WHERE záradék csak a FLOW REPLACE WHERE záradékkal létrehozott streaming táblán támogatott. Ha bármely más streamelési táblában megadja, hibaüzenet jelenik meg. Lásd REFRESH : (MATERIALIZED VIEW vagy STREAMING TABLE).

DML-utasítások

Futtassa a DML-utasításokat közvetlenül a céltáblán a folyamattól eltérő forrásból származó sorok betöltéséhez:

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

Teljes frissítési viselkedés

A CSERE WHERE folyamat teljes frissítése csak a folyamat meghatározott predikátumának használatával hajtja végre újra a forrás lekérdezést. Véglegesen törli a cél minden olyan sorát, amely nem felel meg az adott predikátumnak, beleértve azokat a sorokat is, amelyek felülírják a predikátumot vagy a korábban a meghatározott predikátumtartományon kívülre szúrt DML-utasításokat.

Warning

A teljes frissítés törli az összes meglévő adatot, és csak a meghatározott predikátum használatával hajtja végre újra a folyamatot. Ha egy folyamat egy éve fut egy 7 napos predikátummal, a teljes frissítés csak az utolsó 7 napot tartalmazó táblát eredményezi. Az összes régebbi sor véglegesen törlődik.

REFRESH STREAMING TABLE orders_enriched FULL;

Ha meg szeretné akadályozni egy tábla teljes frissítését, állítsa a táblatulajdonságot pipelines.reset.allowed a következőre false:

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

Növekményes frissítés

A REPLACE WHERE folyamatok lehetőség szerint növekményes frissítést használnak, és a teljes csereablak újrafordítása helyett csak az utolsó frissítés óta megváltozott forrásadatokat dolgozza fel újra. A növekményes frissítés kiszolgáló nélküli számítást igényel.

Növekményes frissítés alkalmazása esetén

Az alábbiak mindegyikének igaznak kell lennie:

• A folyamat kiszolgáló nélküli számításon fut.
• A lekérdezésalak támogatott. Lásd a támogatott operátorkészlet növekményes frissítését .
• A predikátum egy forrástáblából származó alaposzlopokra hivatkozik. A származtatott értékek, például az összesítő vagy az ablakfüggvény kimeneteinek predikátumai nem küldhetők le egy forrásba, amely letiltja a növekményes frissítést.
• Az aktuális csereintervallumban egyetlen külső DML sem módosította a sorokat. Az aktuális ablakon kívüli sorokat módosító DML-t nem érinti.
• Az aktuális csereablak nem tartalmaz olyan sorokat, amelyeket az előző predikátum kizárt. Ha kibővíti a predikátumot egy korábban nem feldolgozott tartomány lefedésére, az egy frissítés vissza fog térni a teljes újraszámításra. A további frissítések ismét jogosultak a növekményes frissítésre.
• A predikátum determinisztikus. A nem determinisztikus függvényeket használó predikátumok, mint például a rand(), letiltják a növekményes frissítést. Az olyan időbeli függvények, mint current_date() az engedélyezettek.

Bármely folyamat első frissítése mindig teljes újraszámítás. Ha bármely feltétel nem teljesül, az adott frissítés visszatér az aktuális csereablak teljes újraszámítására.

Ajánlott eljárások a növekményes frissítéshez

Kövesse ezeket az irányelveket, így a REPLACE-folyamatok WHERE továbbra is jogosultak maradnak a növekményes frissítésre.

Mozgó alsó határ használata

A mozgó alsó határral rendelkező predikátumok korlátlan ideig jogosultak maradnak a növekményes frissítésre.

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

A mozgó felső határ (például date BETWEEN date_add(current_date(), -7) AND current_date()) eltolhatja az ablakot a korábban kizárt sorok belefoglalására, és egyszeri visszalépést válthat ki teljes újraszámításra.

A predikátum oszlop belefoglalása a következőbe: GROUP BY

Összesítéskor vegye fel a predikátumoszlopot a(z) GROUP BY elembe, hogy a motor az aggregáció alá küldhesse le a predikátumot.

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

Ha a predikátumoszlop hiányzik a(z) GROUP BY-ból, a predikátum nem tolható le az aggregáció alá, és a rendszer teljes egészében beolvassa a forrást.

Vegye fel a predikátumoszlopot az összekapcsolási kulcsok közé

Adja hozzá a predikátum oszlopot az illesztési feltételhez, hogy a motor az összes csatlakoztatott forrást meg tudja metszeni.

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;

Ha egy összekapcsolt tábla nem teszi elérhetővé a predikátum oszlopot, a program az egyes frissítéseken teljes egészében beolvasja a táblát.

A teljes újraszámításra való visszaállás diagnosztikája

Ha egy frissítés visszaesik a teljes újraszámításra, az ok a folyamat eseményében planning_information lesz jelentve. Lásd a IssueType objektumot a pipeline eseménynapló sémájában. Az alábbi táblázat az eseményben jelentett okokat sorolja fel:

Reason	Meaning
EXTERNAL_CHANGE_IN_REPLACE_WINDOW	Egy külső DML sorokat módosított az aktuális csereablakban.
REPLACE_WHERE_NOT_DETERMINISTIC	A predikátum nem determinisztikus kifejezéseket használ.
PRIOR_REPLACE_WHERE_NOT_DETERMINISTIC	Az előző frissítés nem determinisztikus predikátumot használt.
PRIOR_REPLACE_WHERE_PREDICATE_MISSING	Az előző frissítésben használt predikátum nem volt visszanyerhető. Ez akkor fordulhat elő, ha a predikátumot nem szerializálták megfelelően a frissítés során.
STALE_TARGET_DATA_IN_WINDOW	A predikátum változott frissítések között, és a jelenlegi ablak tartalmazza azokat a sorokat, amelyeket az előző predikátum kizárt.
CANNOT_PUSH_PREDICATE	A predikátum legalább egy forrásba nem küldhető le.
REPLACE_WHERE_BACKFILL	A frissítés egy visszamenőleges feltöltés, amely egyszeri predicate override-ot használ a korábbi adatok feltöltésére vagy a korábbi írási műveletek javítására.

Examples

Az alábbi példák a REPLACE WHERE folyamatának gyakori mintáit mutatják be.

1. példa: A korábbi összesítések megőrzése korlátozott megőrzési forrásból

Ez a példa határozatlan ideig tartja fenn a napi összesítéseket, még akkor is, ha a nyers adatok a forrástáblán kívülre öregednek (3 napos megőrzés):

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;

2. példa: A dimenziótáblák módosításakor történő újraszámítás megakadályozása

Ez a példa a dimenzióattribútumok változásakor változatlanul hagyja a korábbi ténysorokat:

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;

Ha egy felhasználó régiója megváltozik, csak a legutóbbi sorok lesznek újrafordítve. A historikus sorok megtartják a régió íráskori értékét.

3. példa: Új metrika hozzáadása a teljes előzmények újrafordítása nélkül

Ez a példa bemutatja, hogyan fejleszthet táblázatdefiníciót, és hogyan tölthet vissza csak egy célzott tartományt:

1. Adja meg a kezdeti táblát:

   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. Frissítse a lekérdezést, és adja hozzá ezt: 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;
   

   A 7 naposnál régebbi sorok a következőt NULLtartalmazzákuniq_users: .

4. példa: Iterálás egy kis ablakban a teljes előzmények kitöltése előtt

Ez a példa bemutatja, hogyan érvényesítheti a lekérdezési logikát egy kis adatablakban a teljes előzménytartomány feldolgozása előtt.

Kezdje egy rövid ablakkal a metrikák ellenőrzéséhez és az üzleti logika iterálásához alacsonyabb számítási költségekkel:

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;

Egy rövid időablak minden frissítéskor csak az utolsó 7 napot számolja újra, ezért a teljes történeti futtatás véglegesítése előtt annyiszor módosítsa a lekérdezést, ahányszor szükséges.

A lekérdezés véglegesítése után a DML használatával töltse ki a teljes előzménytartományt:

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;