Dávkové zpracování s toky REPLACE WHERE

Toky REPLACE WHERE v deklarativních kanálech Lakeflow Spark znovu vypočítají a přepíšou určenou podmnožinu tabulky, aniž by bylo nutné znovu zpracovávat 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

Definujte toky REPLACE WHERE v SQL nebo Python.

SQL

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

CREATE STREAMING TABLE orders_enriched
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;

Alternativně použijte syntaxi dlouhého formátu CREATE FLOW :

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;

Python

V Python se tabulka a tok definují v jednom příkazu. Tok dědí stejný název jako tabulka:

from pyspark import pipelines as dp
from pyspark.sql import functions as F
from pyspark.sql.functions import col

@dp.table(
  replace_where=col("date") >= F.date_sub(F.current_date(), 7)
)
def orders_enriched():
  orders_fct = spark.read.table("orders_fct").select("date", "order_id", "region", "qty", "price")
  product_dim = spark.read.table("product_dim")
  return orders_fct.join(product_dim, "product_id")

Parametr replace_where přijímá buď výraz sloupce PySpark, nebo predikát řetězce.

V těchto příkladech se všechny řádky z posledních 7 dnů odstraní z orders_enriched a znovu vypočítají na základě 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 vyžaduje se v SQL. Přiřazuje sloupce podle názvu spíše než podle pozice.

Obnovení historických dat

Pokud chcete do cílové tabulky zapsat historické nebo opravené řádky mimo plánované aktualizace, zvolte mezi dvěma mechanismy na základě toho, kde se nachází historická data:

Přepsání predikátu: Znovu spusťte zdrojový dotaz toku pro jednorázový rozsah predikátu. Používá se, když historická data pocházejí ze stejného zdroje jako přírůstková data.
Příkazy DML: Vložení přímo do cílové tabulky a obejití toku. Používá se, když se historická data nacházejí v jiném zdroji než přírůstková data.

Přepsání predikátů

Přepište predikát REPLACE WHERE pro jednu aktualizaci kanálu beze změny definice kanálu. Přepisy predikátů jsou jednorázové, platí pouze pro aktuální aktualizaci a nemají vliv na budoucí spuštění.

Příklad: Počáteční historické zatížení

Chcete-li při prvním nastavení datového kanálu provést jednorázové zpětné doplnění historických dat:

pipeline_id = "<pipeline-id>"
overrides = [
  {
    "flow_name": "orders_enriched",
    "predicate_override": "date BETWEEN '2020-01-01' AND '2024-12-31'",
  }
]

resp = start_update_with_replace_where(
  pipeline_id=pipeline_id,
  replace_where_overrides=overrides,
)
print(resp)

Příklad: Oprava sloupce pro určité období

Po aktualizaci definice sloupce znovu vyplňte změnu cílového historického rozsahu:

pipeline_id = "<pipeline-id>"
overrides = [
  {
    "flow_name": "orders_enriched",
    "predicate_override": "date >= date_add(current_date(), -30)",
  }
]

resp = start_update_with_replace_where(
  pipeline_id=pipeline_id,
  replace_where_overrides=overrides,
  refresh_selection=["orders_enriched"],
)
print(resp)

Zkombinujte více dimenzí v rámci jednoho přepsání predikátu:

overrides = [
  {
    "flow_name": "orders_enriched",
    "predicate_override": "date >= date_add(current_date(), -30) AND region = 'asia'",
  }
]

Pomocná funkce: start_update_with_replace_where

Pomocí rozhraní API pro aktualizaci kanálu z notebooku odešlete přepsání predikátů:

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.pipelines import StartUpdateResponse


def start_update_with_replace_where(
  pipeline_id: str,
  replace_where_overrides: list[dict],
  refresh_selection: list[str] = None,
) -> StartUpdateResponse:
  """Start a pipeline update with REPLACE WHERE predicate overrides."""
  client = WorkspaceClient()

  body = {
    "pipeline_id": pipeline_id,
    "cause": "JOB_TASK",
    "update_cause_details": {
      "job_details": {"performance_target": "PERFORMANCE"}
    },
    "replace_where_overrides": replace_where_overrides,
  }

  if refresh_selection:
    body["refresh_selection"] = refresh_selection

  res = client.api_client.do(
    "POST",
    f"/api/2.0/pipelines/{pipeline_id}/updates",
    body=body,
    headers={"Accept": "application/json", "Content-Type": "application/json"},
  )

  return StartUpdateResponse.from_dict(res)

Příkazy DML

Spouštějte příkazy DML přímo nad cílovou tabulkou mimo datovou pipeline, abyste provedli počáteční načtení dat nebo opravy, například načtení z tabulky staršího systému:

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

Řádky vložené prostřednictvím DML nejsou předmětem predikátu REPLACE WHERE a uchovávají se napříč plánovanými aktualizacemi, pokud nepřejdou do rozsahu predikátu budoucího spuštění.

Chování úplné aktualizace

Úplná aktualizace toku REPLACE WHERE znovu spustí zdrojový dotaz pouze pomocí aktuálního predikátu. Řádky vložené pomocí přepsání predikátu nebo příkazů 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í.

Chcete-li zabránit úplnému obnovení tabulky, nastavte vlastnost tabulky pipelines.reset.allowed na false. Viz referenční informace o vlastnostech kanálu.

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	Meaning
`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.

Omezení

Toky REPLACE WHERE mají následující omezení:

Cílová tabulka musí být vytvořena v rámci pipeline.
Pro cílovou tabulku je povolený pouze jeden tok REPLACE WHERE .
Na tabulku, na kterou cílí tok typu REPLACE WHERE, nemůže zároveň cílit jiný typ toku, například tok typu AUTO CDC nebo tok přidávání.
U tabulek, na které jsou cíleny toky REPLACE WHERE, nejsou očekávání podporována.
Informace o samostatných streamovacích tabulkách najdete v tématu Toky REPLACE WHERE pro samostatné streamovací tabulky, kde jsou popsány rozdíly v syntaxi a zpětném doplnění dat.

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í:

SQL

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

Python

from pyspark import pipelines as dp
from pyspark.sql import functions as F
from pyspark.sql.functions import col

@dp.table(
  replace_where=col("date") >= F.date_sub(F.current_date(), 3)
)
def events_agg():
  return (
    spark.read.table("events_raw")
      .groupBy("date", "key")
      .agg(F.sum("val").alias("agg"))
  )

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í:

SQL

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

Python

from pyspark import pipelines as dp
from pyspark.sql import functions as F
from pyspark.sql.functions import col

@dp.table(
  replace_where=col("date") >= F.date_sub(F.current_date(), 1)
)
def fact_dim_join():
  fact_table = spark.read.table("fact_table").alias("f")
  dim_users = spark.read.table("dim_users").alias("d")
  return (
    fact_table.join(dim_users, col("f.user_id") == col("d.user_id"))
      .select(
        col("f.date"),
        col("f.user_id"),
        col("d.region"),
        col("f.revenue"),
      )
  )

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. Chcete-li opravit historické záznamy, spusťte cílené doplnění dat pomocí přepsání predikátů.

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:

Definujte počáteční tabulku:

SQL

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

Python

from pyspark import pipelines as dp
from pyspark.sql import functions as F
from pyspark.sql.functions import col

@dp.table(
  replace_where=col("event_date") >= F.date_sub(F.current_date(), 7)
)
def clickstream_daily():
  return (
    spark.read.table("clickstream_raw")
      .groupBy("event_date", "page_id")
      .agg(F.count("*").alias("clicks"))
  )

Aktualizujte dotaz tak, aby přidal uniq_users:

SQL

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

Python

@dp.table(
  replace_where=col("event_date") >= F.date_sub(F.current_date(), 7)
)
def clickstream_daily():
  return (
    spark.read.table("clickstream_raw")
      .groupBy("event_date", "page_id")
      .agg(
        F.count("*").alias("clicks"),
        F.countDistinct("user_id").alias("uniq_users"),
      )
  )

Zpětně doplňte novou metriku za posledních 30 dnů:

overrides = [
  {
    "flow_name": "clickstream_daily",
    "predicate_override": "event_date BETWEEN '2026-01-01' AND '2026-01-30'",
  }
]

resp = start_update_with_replace_where(
  pipeline_id="<pipeline-id>",
  replace_where_overrides=overrides,
  refresh_selection=["clickstream_daily"],
)

Řádky starší než zpětně doplněný rozsah obsahují 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, aby každá aktualizace přepočítává pouze posledních 7 dnů při revizi dotazu:

SQL

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

Python

from pyspark import pipelines as dp
from pyspark.sql import functions as F
from pyspark.sql.functions import col

@dp.table(
  replace_where=col("event_date") >= F.date_sub(F.current_date(), 7)
)
def revenue_attribution():
  return (
    spark.read.table("marketing_events")
      .groupBy("event_date", "campaign_id")
      .agg(F.sum("revenue").alias("total_revenue"))
  )

Po dokončení dotazu použijte přepsání predikátu k provedení jednorázového zpětného doplnění historických dat:

overrides = [
  {
    "flow_name": "revenue_attribution",
    "predicate_override": "event_date >= date_add(current_date(), -365)",
  }
]

resp = start_update_with_replace_where(
  pipeline_id="<pipeline-id>",
  replace_where_overrides=overrides,
  refresh_selection=["revenue_attribution"],
)

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-07-07

Dávkové zpracování s toky REPLACE WHERE

Requirements

Kdy použít toky typu REPLACE WHERE

Vytvoření toku REPLACE WHERE

SQL

Python

Obnovení historických dat

Přepsání predikátů

Příklad: Počáteční historické zatížení

Příklad: Oprava sloupce pro určité období

Příkazy DML

Chování úplné aktualizace

Přírůstková aktualizace

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

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

Použijte pohyblivou dolní mez

Zahrňte sloupec predikátů do GROUP BY

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

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

Omezení

Příklady

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

SQL

Python

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

SQL

Python

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

SQL

Python

SQL

Python

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

SQL

Python

Váš názor

Další materiály