Använd Python med fristående pipelines

Important

Att skapa och uppdatera fristående materialiserade vyer och strömmande tabeller från en notebook på serverlös generell beräkning är i Beta och tillgängligt i utvalda regioner. Se Anteckningsböcker.

Du kan skapa och uppdatera fristående materialiserade vyer och strömmande tabeller från en notebook-fil med hjälp av Python. På så sätt kan du hantera fristående pipelines tillsammans med dina andra Python-baserade notebook-arbetsflöden.

Det finns två sätt att göra detta:

  • Definiera bordet med dekoratörerna pyspark.pipelines , @dp.materialized_view och @dp.table. Använd detta när logiken är lättare att uttrycka som DataFrame-kod. Se Definiera tabeller med dekoratörer för pipelines.
  • Skicka in samma SQL-satser som Databricks SQL-lager kör genom att skicka dem till spark.sql(). Detta ger dig fullständigt SQL-stöd för fristående materialiserade vyer och strömmande tabeller, inklusive REFRESH-satser och uppdateringsscheman. Se Skicka SQL-instruktioner med spark.sql().

Python-källkod för fristående pipelines kräver en notebook som är ansluten till serverlös generell beräkning. Du kan inte använda Python för att skapa eller uppdatera fristående pipelines från ett Databricks SQL-lager eftersom ett lager kör SQL-instruktioner, inte Python notebook-filer. Information om hur du använder ett SQL-lager finns i Använda fristående materialiserade vyer och Använd fristående strömningstabeller.

Requirements

Om du vill skapa och uppdatera fristående pipelines med Python behöver du en notebook som är ansluten till serverlös generell beräkning i Databricks Runtime 18.1 eller senare. En fullständig lista över krav, inklusive regional tillgänglighet och behörigheter, finns i Notebook-filer.

Definiera tabeller med pipeline-dekoratörerna

Du kan definiera en fristående materialiserad vy eller strömningstabell med samma dekoratörer som du använder i en Lakeflow-pipeline. Varje dekorerad funktion definierar en tabell. När du kör cellen skapar Azure Databricks tabellen och kör en serverlös pipeline för att fylla den. Cellen visas igen när uppdateringen har slutförts.

Varning

Pipeline-dekoratörer kräver version av serverlös miljö 5 eller senare.

Definiera en materialiserad vy

Använd @dp.materialized_view på en funktion som returnerar en batch-DataFrame. Följande exempel skapar den materialiserade vyn daily_booking_revenue från bookings tabellen i Wanderbricks exempeldataset:

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"))
  )

För att definiera en tabell från en strömmande läsning, använd @dp.table istället.

Definiera en strömmande tabell

Använd @dp.table på en funktion som returnerar en strömmande DataFrame. Följande exempel skapar streamingtabellen bookings_raw från en strömmande läsning av samma bookings-tabell:

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")

Om funktionen returnerar en batch DataFrame, @dp.table skapar istället en materialiserad vy. Det enda undantaget är replace_where, vilket alltid resulterar i en strömningstabell. Följande exempel håller dagliga intäkter för incheckningar på eller efter den 1 juli 2025 aktuella, utan att beräkna om tidigare datum:

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"))
  )

Varje körning tar bort raderna som matchar predikatet och beräknar om endast det intervallet. Se batchbearbetning med REPLACE WHERE-flöden.

Uppdatera en tabell

För att uppdatera en tabell du definierat med en dekoratör, kör koden som definierar den igen, till exempel genom att köra om notebookcellen, köra hela notebooken eller köra notebooken som ett jobb. Varje körning skapar tabellen om den inte finns och uppdaterar den om den redan finns.

För att återbearbeta all tillgänglig data i källan, skicka full_refresh=True vidare till någon av dekoratörerna:

@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")

Du kan inte använda ett REFRESH statement på en tabell definierad med en dekoratör, eller schemalägga uppdateringar med SCHEDULE eller TRIGGER ON UPDATE. För att uppdatera ett schema, definiera tabellen i SQL eller schemalägg anteckningsboken som ett jobb. Se Lakeflow Jobs.

Konfigurera tabellen

Dekoratörerna accepterar samma vanliga datasetparametrar som de accepterar i en pipeline, inklusive comment, table_properties, , partition_cols, cluster_by, schema, och 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"))
  )

För parameterlistan, se materialized_view och tabell.

private=True stöds inte, eftersom en privat tabell endast kan läsas av andra dataset i samma pipeline.

API:er som inte stöds

En fristående tabell är en enda datamängd med ett enda flöde, så API:erna som beskriver relationer mellan dataset är inte tillgängliga. Följande ger ett fel utanför en pipeline:

  • @dp.temporary_view och dp.create_streaming_table
  • @dp.append_flow och andra ytterligare flöden
  • dp.create_auto_cdc_flow och dp.create_auto_cdc_from_snapshot_flow
  • @dp.replace_flow och parametern replace_using, som definierar REPLACE USING-flöden. Se Partiell ögonblicksbildsersättning med REPLACE USING-flöden.
  • dp.create_sink
  • Förväntningar, såsom @dp.expect och @dp.expect_or_fail

För att använda dessa, skapa istället en Lakeflow-pipeline. Se utveckla pipelinekod med Python.

Skicka in SQL-satser med spark.sql()

I en Python-anteckningsbok skickar du samma satser som du skulle köra från ett Databricks SQL Warehouse till spark.sql(). Syntaxen för fristående materialiserade vyer och strömningstabeller är identisk; endast sättet du skickar instruktionen på skiljer sig åt. Precis som med ett datalager kör varje CREATE- eller REFRESH-instruktion en serverlös pipeline för att bearbeta operationen.

Sessionen spark är tillgänglig som standard i Azure Databricks notebook-filer, så ingen import krävs.

Skapa en materialiserad vy

I följande exempel skapas den materialiserade vyn mv1 från bastabellen 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
""")

Fullständig CREATE MATERIALIZED VIEW information, till exempel schemalagda och utlösta uppdateringar, finns i Skapa en materialiserad vy.

Skapa en direktuppspelningstabell

I följande exempel skapas strömningstabellen salesraw_data från tabellen:

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

Fullständig CREATE STREAMING TABLE information, inklusive inläsning av filer med Auto Loader och schemaläggning, finns i Använda fristående strömningstabeller.

Uppdatera en materialiserad vy eller en strömmande tabell

Använd en REFRESH instruktion för att uppdatera en fristående tabell med de senaste data från källan:

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

Vid serverlös allmän beräkning är uppdateringar synkrona. Asynkrona uppdateringar (nyckelordet ASYNC ) stöds inte. Se Serverlös allmän beräkning.

Parametrisera satser

Om du vill skicka värden från din Python kod till en instruktion i stället för att hårdkoda dem använder du namngivna parametermarkörer i SQL och anger deras värden via args argumentet spark.sql(). Använd en markör, till exempel :min_sales direkt för literalvärden. Radbryt markören IDENTIFIER() endast när parametern är ett objektnamn, till exempel en tabell, vy eller ett schema, eftersom identifierare inte kan ersättas med vanliga strängvärden.

I följande exempel parameteriseras både det materialiserade vynamnet och ett filtervärde:

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,
})

Mer information finns i Parametermarkörer och IDENTIFIER -sats.

Kör andra instruktioner

Du kan köra valfri fristående materialiserad vy eller strömmande tabellinstruktion från en Python notebook-fil genom att skicka den till spark.sql(), inklusive instruktioner för att schemalägga uppdateringar, ändra en tabell eller släppa en tabell. Information om hur du använder materialiserade vyer och strömmande tabeller, inklusive SQL-syntax, finns i Använda fristående materialiserade vyer och Använd fristående strömningstabeller.

Limitations

Fristående materialiserade vyer och strömmande tabeller som skapats på serverlös allmän beräkning har ytterligare begränsningar, till exempel inget stöd för asynkrona uppdateringar och ingen kostnadsattribution per tabell. Den fullständiga listan finns i Serverlös allmän beräkning.

Eftersom dessa pipelines körs på serverlös generell beräkning i stället för ett SQL-datalager ärver de inte anpassade taggar från ett överordnat datalager. Taggpropagering från lager till system.billing.usage gäller endast materialiserade vyer och streamingtabeller vars instruktioner körs från ett SQL-datalager. Se Attributkostnader till SQL-lagret med anpassade taggar.

Tabeller definierade med pipeline-dekoratörerna har följande ytterligare begränsningar:

  • Du kan inte uppdatera dem med en REFRESH-sats eller schemalägga uppdatering med SCHEDULE eller TRIGGER ON UPDATE. Se Uppdatera en tabell.
  • Förväntningar, tilläggsflöden, ändringsdatafångst (CDC), sänkor och tillfälliga vyer stöds inte. Se Ej stödda API:er.
  • private=True stöds inte.

Ytterligare resurser

  • Uppdaterades senast