Python használata különálló folyamatláncokkal

Important

Az önálló materializált nézetek és streamelési táblák létrehozása és frissítése egy jegyzetfüzetből a kiszolgáló nélküli általános számításban bétaverzióban érhető el, és bizonyos régiókban érhető el. Lásd : Jegyzetfüzetek.

Különálló materializált nézeteket és streamelő táblákat hozhat létre és frissíthet egy jegyzetfüzetből Python használatával. Így a többi Python-alapú jegyzetfüzet-munkafolyamat mellett önálló folyamatokat is kezelhet.

Ezt kétféleképpen teheti meg:

  • Definiáljuk a táblát a pyspark.pipelines dekorátorokkal @dp.materialized_view , és @dp.table. Használd ezt, amikor a logika könnyebben kifejezhető DataFrame kódként. Lásd: Táblák definiálása pipeline-dekorátorokkal.
  • Küldje el ugyanazokat az SQL-utasításokat, amelyeket a Databricks SQL-warehouse futtat, úgy, hogy átadja őket a(z) spark.sql() számára. Ez biztosítja az önálló materializált nézetek és a streamelési táblák teljes SQL-felületét, beleértve a REFRESH utasításokat és a frissítési ütemezéseket. Lásd SQL-utasítások elküldése a(z) spark.sql() használatával.

Az önálló folyamatok Python-forrásához egy, a kiszolgáló nélküli általános számítási környezethez csatlakoztatott notebook szükséges. A Python nem használható önálló folyamatok létrehozására vagy frissítésére a Databricks SQL Warehouse-ból, mert egy raktár SQL-utasításokat futtat, nem Python jegyzetfüzeteket. Ha inkább SQL Warehouse-t szeretne használni, olvassa el az önálló materializált nézetek ésaz önálló streamelési táblák használata című témakört.

Requirements

Az önálló adatfolyamok Pythonnal történő létrehozásához és frissítéséhez egy, a Databricks Runtime 18.1-es vagy újabb verzióján futó, szerver nélküli általános számítási erőforráshoz csatlakoztatott jegyzetfüzetre van szükség. A követelmények teljes listájáért, beleértve a regionális rendelkezésre állást és az engedélyeket, olvassa el a Jegyzetfüzetek című témakört.

Táblák definiálása pipeline-dekorátorokkal

Ugyanazokkal a dekorátorokkal definiálhatsz önálló materializált nézetet vagy streaming táblát, amelyeket egy Lakeflow-folyamatban használsz. Minden dekorált függvény egy táblát definiál. Amikor futtatod a cellát, az Azure Databricks létrehozza a táblát, és egy szerver nélküli pipeline-t futtat annak feltöltésére. A cella újra megjelenik, amikor a frissítés befejeződik.

Warning

A folyamatdekorátorokhoz a kiszolgáló nélküli környezet 5-ös vagy újabb verziója szükséges.

Definiáljuk a materializált nézetet

Használd @dp.materialized_view egy olyan függvényen, amely egy batch DataFrame-et ad vissza. A következő példa létrehozza a Wanderbricks mintaadathalmazbookings táblájából a daily_booking_revenue materializált nézetet:

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

@dp.materialized_view(name="main.default.daily_booking_revenue")
def daily_booking_revenue():
  return (
    spark.read.table("samples.wanderbricks.bookings")
    .groupBy("check_in")
    .agg(F.sum("total_amount").alias("total_revenue"))
  )

Egy táblázat meghatározásához egy streaming olvasásból használd @dp.table helyette.

Definiáljunk egy streaming táblát

Használd @dp.table egy olyan függvényen, amely streaming DataFrame-et ad vissza. Az alábbi példa hozza létre a streaming táblát bookings_raw ugyanannak bookings a táblázatnak a streaming olvasatából:

from pyspark import pipelines as dp

@dp.table(name="main.default.bookings_raw")
def bookings_raw():
  return spark.readStream.table("samples.wanderbricks.bookings").select("booking_id", "property_id", "total_amount")

Ha a függvény egy köteg DataFrame-t ad vissza, @dp.table helyette materializált nézetet hoz létre. Az egyetlen kivétel , replace_whereami mindig egy streaming táblát eredményez. Az alábbi példa naprakész tartja a napi bevételeket a 2025. július 1-jén vagy azt követő bejelentkezésekhez, anélkül, hogy újra kiszámolná a korábbi dátumokat:

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

@dp.table(
  name="main.default.booking_revenue_rw",
  replace_where=F.col("check_in") >= F.to_date(F.lit("2025-07-01")),
)
def booking_revenue_rw():
  return (
    spark.read.table("samples.wanderbricks.bookings")
    .groupBy("check_in")
    .agg(F.sum("total_amount").alias("total_revenue"))
  )

Minden futás törli a predikátumhoz tartozó sorokat, és csak azt a tartományt számolja újra. Lásd: Kötegelt feldolgozás REPLACE WHERE folyamatokkal.

Táblázat frissítése

Egy dekorátorral meghatározott táblázat frissítéséhez újra futtatjuk a definiáló kódot, például újrafuttatva a notebook cellát, az egész jegyzetfüzetet, vagy a jegyzetfüzetet munkaként futtatva. Minden futás létrehozza a táblázatot, ha nem létezik, és frissíti, ha létezik.

A forrásban elérhető összes adat újrafeldolgozásához adja át a(z) full_refresh=True értéket valamelyik dekorátornak:

@dp.table(name="main.default.bookings_raw", full_refresh=True)
def bookings_raw():
  return spark.readStream.table("samples.wanderbricks.bookings").select("booking_id", "property_id", "total_amount")

Nem használhatsz REFRESH utasítást dekorátorral definiált táblán, és nem ütemezhetsz frissítéseket az SCHEDULE vagy a TRIGGER ON UPDATE használatával. Az ütemezés frissítéséhez definiáld a táblát SQL-ben, vagy ütemezd a jegyzetfüzetet munkaként. Lásd Lakeflow Jobs.

Konfiguráld a táblázatot

A dekorátorok ugyanazokat a közös adathalmazparamétereket fogadják el, mint amelyeket egy csővezetéken belül elfogadnak, beleértve comment, table_properties, partition_cols, cluster_byschema, és spark_conf:

@dp.materialized_view(
  name="main.default.daily_booking_revenue",
  comment="Daily booking revenue.",
  table_properties={"quality": "gold"},
  cluster_by=["check_in"],
)
def daily_booking_revenue():
  return (
    spark.read.table("samples.wanderbricks.bookings")
    .groupBy("check_in")
    .agg(F.sum("total_amount").alias("total_revenue"))
  )

A paraméterlistáról lásd a materialized_view és a táblázatot.

private=True nem támogatott, mert egy privát táblát csak más adathalmazok olvashatnak ugyanabban a csővezetékben.

Nem támogatott API-k

Az önálló tábla egyetlen adathalmaz, amelynek egyetlen áramlása van, így az adathalmazok közötti kapcsolatokat leíró API-k nem elérhetők. Az alábbiak hibát okoznak a csővezetéken kívül:

  • @dp.temporary_view és dp.create_streaming_table
  • @dp.append_flow és egyéb további áramlások
  • dp.create_auto_cdc_flow és dp.create_auto_cdc_from_snapshot_flow
  • @dp.replace_flow és a replace_using paraméter, amely a REPLACE USING flow-okat definiálja. Lásd Részleges pillanatképcsere a REPLACE USING folyamatokkal.
  • dp.create_sink
  • Elvárások, mint @dp.expect és @dp.expect_or_fail

Ezek használatához ehelyett hozzon létre egy Lakeflow-folyamatot. Lásd: Folyamatkód fejlesztése Python-használatával.

SQL utasítások küldése spark.sql()

Egy Python-notebookban adja meg ugyanazokat az utasításokat, amelyeket egy Databricks SQL-warehouse-ból futtatna a spark.sql() számára. Az önálló materializált nézet és a streamelési táblázat szintaxisa megegyezik; csak az utasítás elküldésének módja különbözik. A raktárhoz hasonlóan mindegyik CREATE vagy REFRESH utasítás kiszolgáló nélküli folyamatot futtat a művelet feldolgozásához.

A spark munkamenet alapértelmezés szerint Azure Databricks jegyzetfüzetekben érhető el, ezért nincs szükség importálásra.

Materializált nézet létrehozása

Az alábbi példa létrehozza a materializált nézetet mv1 az alaptáblából base_table1:

spark.sql("""
  CREATE OR REPLACE MATERIALIZED VIEW mv1
  AS SELECT
    date,
    sum(sales) AS sum_of_sales
  FROM base_table1
  GROUP BY date
""")

A teljes CREATE MATERIALIZED VIEW részletekért, például az ütemezett és eseményindítású frissítésekről, lásd a következőt: Materializált nézet létrehozása.

Streaming tábla létrehozása

Az alábbi példa létrehozza a streamelési táblát sales a raw_data táblából:

spark.sql("""
  CREATE OR REFRESH STREAMING TABLE sales
  AS SELECT product, price FROM STREAM raw_data
""")

További részletekért CREATE STREAMING TABLE , beleértve a fájlok automatikus betöltővel való betöltését és az ütemezést, olvassa el az önálló streamelőtáblák használatát ismertető cikket.

Materializált nézet vagy streamingtábla frissítése

REFRESH Egy utasítással frissíthet egy különálló táblát a forrásból származó legfrissebb adatokkal:

spark.sql("REFRESH MATERIALIZED VIEW mv1")
spark.sql("REFRESH STREAMING TABLE sales")

A kiszolgáló nélküli általános számításban a frissítések szinkronban vannak. Az aszinkron frissítések (a ASYNC kulcsszó) nem támogatottak. Lásd : Kiszolgáló nélküli általános számítás.

Utasítások paraméterezése

Ha a Python-kódból szeretne értékeket egy utasításnak átadni ahelyett, hogy fixen beírná őket, használjon elnevezett paraméterjelölőket az SQL-ben, és az értékeiket a argsspark.sql() argumentumán keresztül adja meg. Használjon jelölőt, például :min_sales közvetlenül a literális értékekhez. A jelölőt csak akkor burkolja be IDENTIFIER() , ha a paraméter egy objektumnév, például tábla, nézet vagy séma, mert az azonosítók nem helyettesíthetők egyszerű sztringértékekként.

Az alábbi példa paraméterezi a materializált nézet nevét és a szűrő értékét is:

mv_name = "main.sales.regional_sales"
min_sales = 1000

spark.sql("""
  CREATE OR REPLACE MATERIALIZED VIEW IDENTIFIER(:mv)
  AS SELECT
    region,
    sum(sales) AS sum_of_sales
  FROM base_table1
  WHERE sales > :min_sales
  GROUP BY region
""", args={
  "mv": mv_name,
  "min_sales": min_sales,
})

További információ: Paraméterjelölők és IDENTIFIER záradék.

Egyéb utasítások futtatása

Bármely különálló materializált nézetet vagy streamelési táblázatutasítást futtathat egy Python jegyzetfüzetből, ha átadja azt a jegyzetfüzetnekspark.sql(), beleértve a frissítések ütemezésére, a táblák módosítására vagy a táblák elvetésére vonatkozó utasításokat. A materializált nézetek és a streamelőtáblák ( beleértve az SQL szintaxisát) használatának megismeréséhez tekintse meg az önálló materializált nézetek és az önálló streamelési táblák használata című témakört.

Limitations

A kiszolgáló nélküli általános számításon létrehozott önálló materializált nézetek és streamtáblák további korlátozásokkal rendelkeznek, például nem támogatják az aszinkron frissítéseket, és nem rendelhetők táblaenkénti költségek. A teljes listát a Kiszolgáló nélküli általános számítás című témakörben találja.

Mivel ezek a pipeline-ek szerver nélküli általános számítási rendszeren futnak, nem SQL raktárban, nem örökölnek egyedi címkéket egy záró raktárból. Az adattárházcímkék propagálása a(z) system.billing.usage elemre csak azokra a materializált nézetekre és streamelési táblákra vonatkozik, amelyek utasításai SQL-adattárházból futnak. Tekintse meg az SQL-raktár attribútumköltségét egyéni címkékkel.

A pipeline-dekorátorokkal definiált táblákra a következő további korlátozások vonatkoznak:

  • Nem frissítheti őket REFRESH utasítással, és nem ütemezheti a frissítéseket a SCHEDULE vagy TRIGGER ON UPDATE használatával. Lásd : Frissítsd fel a táblázatot.
  • Az elvárások, további áramlások, változásadat-rögzítési (CDC) áramlások, elnyelések és ideiglenes nézetek nem támogatottak. Lásd: Nem támogatott API-k.
  • private=True nem támogatott.

További erőforrások

  • Utolsó frissítés időpontja: