Python gebruiken met zelfstandige pijplijnen

Important

Het maken en vernieuwen van zelfstandige materialized views en streamingtabellen vanuit een notebook op serverloze compute voor algemene doeleinden is bèta en beschikbaar in geselecteerde regio's. Zie Notebooks.

U kunt zelfstandige gerealiseerde weergaven en streamingtabellen maken en vernieuwen vanuit een notebook met behulp van Python. Hiermee kunt u zelfstandige pijplijnen naast uw andere Python notebookwerkstromen beheren.

Er zijn twee manieren om dit te doen:

  • Definieer de tabel met de pyspark.pipelines decorators, @dp.materialized_view en @dp.table. Gebruik dit wanneer de logica makkelijker uit te drukken is als DataFrame-code. Zie Definieer tabellen met de pipelines-decorators.
  • Verzend dezelfde SQL-instructies die door een Databricks SQL-warehouse worden uitgevoerd door ze door te geven aan spark.sql(). Dit geeft je het volledige SQL-oppervlak van de zelfstandige gematerialiseerde weergave en streamingtabel, inclusief REFRESH-instructies en verversingsschema's. Zie SQL-instructies indienen met spark.sql().

Python-broncode voor zelfstandige pijplijnen vereist een notebook dat is gekoppeld aan serverloze algemene rekenkracht. U kunt Python niet gebruiken om zelfstandige pijplijnen te maken of te vernieuwen vanuit een Databricks SQL-warehouse, omdat een warehouse SQL-instructies uitvoert, niet Python notebooks. Als u in plaats daarvan een SQL-warehouse wilt gebruiken, zie Zelfstandige gematerialiseerde weergaven gebruiken en Zelfstandige streamingtabellen gebruiken.

Requirements

Als u zelfstandige pijplijnen wilt maken en vernieuwen met Python, hebt u een notebook nodig dat is gekoppeld aan serverloze algemene berekeningen op Databricks Runtime 18.1 of hoger. Zie Notebooks voor de volledige lijst met vereisten, waaronder regionale beschikbaarheid en machtigingen.

Definieer tabellen met de decorators voor pijplijnen

Je kunt een standalone gematerialiseerde weergave of streamingtabel definiëren met dezelfde decorators die je gebruikt in een Lakeflow-pipeline. Elke gedecoreerde functie definieert één tabel. Wanneer je de cel uitvoert, maakt Azure Databricks de tabel aan en voert een serverloze pipeline uit om deze te vullen. De cel komt terug zodra de update is voltooid.

Warning

De pijplijndecorators vereisen versie van de serverloze omgeving 5 of hoger.

Definieer een gematerialiseerde visie

Gebruik @dp.materialized_view op een functie die een batch DataFrame teruggeeft. Het volgende voorbeeld creëert de gematerialiseerde weergave daily_booking_revenue uit de bookings tabel in de Wanderbricks-voorbeelddataset:

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

Gebruik in plaats daarvan @dp.table om een tabel te definiëren op basis van een streaming read.

Definieer een streamingtabel

Gebruik @dp.table op een functie die een streaming DataFrame teruggeeft. Het volgende voorbeeld creëert de streamingtabel bookings_raw uit een streaming read van dezelfde bookings tabel:

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

Als de functie een batch DataFrame retourneert, maakt @dp.table in plaats daarvan een materialized view. De enige uitzondering is replace_where, wat altijd resulteert in een streamingtabel. Het volgende voorbeeld houdt de dagelijkse opbrengst voor check-ins op of na 1 juli 2025 up-to-date, zonder eerdere data opnieuw te rekenen:

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

Elke run verwijdert de rijen die overeenkomen met het predicaat en berekent alleen dat bereik opnieuw. Zie Batchverwerking met REPLACE-flowsWHERE.

Ververs een tabel

Om een tabel die je met een decorator hebt gedefinieerd te verversen, voer je de code die het definieert opnieuw uit, bijvoorbeeld door de notebookcel opnieuw uit te voeren, het hele notebook te draaien, of het notebook als een taak te laten draaien. Elke run maakt de tabel aan als die niet bestaat en ververst deze als die er wel is.

Om alle gegevens die beschikbaar zijn in de bron opnieuw te verwerken, geef je full_refresh=True mee aan een van beide decorators:

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

Je kunt geen REFRESH statement gebruiken op een tabel die met een decorator is gedefinieerd, of verversingen plannen met SCHEDULE of TRIGGER ON UPDATE. Om volgens een schema te verversen, definieer je de tabel in SQL of plan je het notebook in als taak. Zie Lakeflow Jobs.

Configureer de tabel

De decorators accepteren dezelfde gemeenschappelijke datasetparameters die zij in een pijplijn accepteren, waaronder comment, table_properties, , partition_cols, cluster_by, schema, en 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"))
  )

Voor de parameterlijst, zie materialized_view en tabel.

private=True wordt niet ondersteund, omdat een private tabel alleen door andere datasets in dezelfde pijplijn kan worden gelezen.

Niet-ondersteunde API's

Een standalone tabel is één dataset met één flow, dus de API's die relaties tussen datasets beschrijven zijn niet beschikbaar. De volgende veroorzaken een fout buiten een pijplijn:

  • @dp.temporary_view en dp.create_streaming_table
  • @dp.append_flow en andere aanvullende stromen
  • dp.create_auto_cdc_flow en dp.create_auto_cdc_from_snapshot_flow
  • @dp.replace_flow en de parameter replace_using, die REPLACE USING-stromen definiëren. Zie Vervanging van partiële snapshots met REPLACE USING-flows.
  • dp.create_sink
  • Verwachtingen, zoals @dp.expect en @dp.expect_or_fail

Om deze te gebruiken, kun je in plaats daarvan een Lakeflow-pijpleiding ontwerpen. Zie Pijplijncode ontwikkelen met Python.

Voer SQL-instructies uit met spark.sql()

Geef in een Python notebook dezelfde instructies door die u zou uitvoeren vanuit een Databricks SQL Warehouse naar spark.sql(). De syntaxis van de zelfstandige gematerialiseerde weergave en de streamingtabel is identiek; alleen de manier waarop u de opdracht indient, verschilt. Net als bij een magazijn voert elke CREATE of REFRESH instructie een serverloze pijplijn uit om de bewerking te verwerken.

De spark sessie is standaard beschikbaar in Azure Databricks notebooks, dus er is geen import vereist.

Een gerealiseerde weergave maken

In het volgende voorbeeld wordt de gerealiseerde weergave gemaakt mv1 uit de basistabel 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
""")

Zie CREATE MATERIALIZED VIEW voor volledige details, zoals geplande en geactiveerde vernieuwingen.

Een streamingtabel maken

In het volgende voorbeeld wordt de streamingtabel sales gemaakt op basis van de raw_data tabel:

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

Zie CREATE STREAMING TABLE voor meer informatie, waaronder het laden van bestanden met automatisch laden en plannen.

Een gematerialiseerde weergave of streamingtabel vernieuwen

Gebruik een REFRESH instructie om een zelfstandige tabel bij te werken met de meest recente gegevens uit de bron:

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

Op serverloze algemene berekeningen zijn vernieuwingen synchroon. Asynchrone vernieuwingen (het ASYNC trefwoord) worden niet ondersteund. Zie Serverloze algemene rekenkracht.

Instructies voor parameteriseren

Als u waarden uit uw Python code wilt doorgeven aan een instructie in plaats van ze te coderen, gebruikt u benoemde parametermarkeringen in de SQL en geeft u hun waarden op via het args argument van spark.sql(). Gebruik een markering, zoals :min_sales rechtstreeks voor letterlijke waarden. Verpakt de markering alleen in IDENTIFIER() wanneer de parameter een objectnaam is, zoals een tabel, weergave of schema, omdat id's niet kunnen worden vervangen als tekenreekswaarden zonder opmaak.

In het volgende voorbeeld worden zowel de gerealiseerde weergavenaam als een filterwaarde geparameteriseerd:

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

Raadpleeg parametermarkeringen en IDENTIFIER-component voor meer informatie.

Andere instructies uitvoeren

U kunt elke zelfstandige gerealiseerde weergave- of streamingtabelinstructie uitvoeren vanuit een Python notebook door deze door te geven aanspark.sql(), inclusief instructies voor het plannen van vernieuwingen, het wijzigen van een tabel of het verwijderen van een tabel. Raadpleeg Zelfstandige gematerialiseerde weergaven gebruiken en Zelfstandige streamingtabellen gebruiken voor informatie over het gebruik van gematerialiseerde weergaven en streamingtabellen, inclusief SQL-syntaxis.

Limitations

Zelfstandige gerealiseerde weergaven en streamingtabellen die zijn gemaakt op serverloze algemene berekeningen hebben extra beperkingen, zoals geen ondersteuning voor asynchrone vernieuwingen en geen kostentoewijzing per tabel. Zie Serverloze algemene berekeningen voor de volledige lijst.

Omdat deze pijplijnen draaien op serverless algemene compute in plaats van op een SQL-warehouse, erven ze geen aangepaste tags van een omsluitend warehouse. Magazijntagpropagatie naar system.billing.usage geldt alleen voor gematerialiseerde views en streamingtabellen waarvan de instructies vanuit een SQL-warehouse draaien. Bekijk de kenmerkkosten voor het SQL-warehouse met aangepaste tags.

Tabellen die met de pipeline-decorators zijn gedefinieerd, hebben de volgende extra beperkingen:

  • Je kunt ze niet verversen met een REFRESH-instructie of verversingen plannen met SCHEDULE of TRIGGER ON UPDATE. Zie Een tabel verversen.
  • Verwachtingen, extra stromen, change data capture (CDC) stromen, sinks en tijdelijke weergaven worden niet ondersteund. Zie Niet-ondersteunde API's.
  • private=True wordt niet ondersteund.

Aanvullende bronnen

  • Laatst bijgewerkt om