Verwenden von Python mit eigenständigen Pipelines

Von Bedeutung

Das Erstellen und Aktualisieren eigenständiger materialisierter Ansichten und Streamingtabellen in einem Notebook auf serverloser Allzweckrechenleistung ist in Beta und in ausgewählten Regionen verfügbar. Siehe Notizbücher.

Sie können eigenständige materialisierte Ansichten und Streamingtabellen aus einem Notizbuch mithilfe von Python erstellen und aktualisieren. Auf diese Weise können Sie eigenständige Pipelines zusammen mit Ihren anderen Python-basierten Notizbuchworkflows verwalten.

Hierfür gibt es zwei Möglichkeiten:

  • Definieren Sie die Tabelle mit den pyspark.pipelines Dekoratoren, @dp.materialized_view und @dp.table. Verwenden Sie dies, wenn die Logik leichter als DataFrame-Code auszudrücken ist. Siehe Tabellen mithilfe von Pipeline-Dekoratoren definieren.
  • Reiche dieselben SQL-Anweisungen ein, die ein Databricks SQL Warehouse ausführt, indem du sie an spark.sql()übergebe. Damit erhalten Sie die vollständige SQL-Oberfläche für eigenständige materialisierte Sichten und Streaming-Tabellen, einschließlich REFRESH-Anweisungen und Aktualisierungszeitplänen. Siehe SQL-Anweisungen mit spark.sql() senden.

Python-Quellcode für eigenständige Pipelines erfordert ein Notebook, das an serverloses allgemeines Computing angehängt ist. Sie können Python nicht verwenden, um eigenständige Pipelines aus einem Databricks SQL Warehouse zu erstellen oder zu aktualisieren, da ein Warehouse SQL-Anweisungen ausführt, nicht Python Notizbücher. Wenn Sie stattdessen ein SQL-Warehouse verwenden möchten, lesen Sie die Abschnitte Verwendung eigenständiger materialisierter Ansichten und Verwendung eigenständiger Streaming-Tabellen.

Requirements

Um eigenständige Pipelines mit Python zu erstellen und zu aktualisieren, benötigen Sie ein Notizbuch, das mit serverlosen allgemeinen Berechnungen auf Databricks Runtime 18.1 oder höher verbunden ist. Die vollständige Liste der Anforderungen, einschließlich regionaler Verfügbarkeit und Berechtigungen, finden Sie unter "Notizbücher".

Definiere Tabellen mit den Pipeline-Dekoratoren

Du kannst eine eigenständige materialisierte Ansicht oder eine Streaming-Tabelle mit denselben Dekorateuren definieren, die du in einer Lakeflow-Pipeline verwendest. Jede dekorierte Funktion definiert einen Tisch. Wenn du die Zelle ausführst, erstellt Azure Databricks die Tabelle und führt eine serverlose Pipeline aus, um sie zu füllen. Die Zelle kehrt zurück, wenn das Update abgeschlossen ist.

Warning

Die Pipeline-Dekoratoren benötigen Version der serverlosen Umgebung 5 oder höher.

Definieren Sie eine materialisierte Sichtweise

Verwenden Sie @dp.materialized_view bei einer Funktion, die einen Batch-DataFrame zurückgibt. Das folgende Beispiel erstellt die materialisierte Ansicht daily_booking_revenue aus der bookings Tabelle im Beispieldatensatz „Wanderbricks“:

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

Um eine Tabelle aus einem Streaming-Lesevorgang zu definieren, verwenden Sie stattdessen @dp.table.

Definiere eine Streaming-Tabelle

Verwenden Sie @dp.table für eine Funktion, die einen Streaming-DataFrame zurückgibt. Das folgende Beispiel erzeugt die Streaming-Tabelle bookings_raw aus einer Streaming-Lektüre derselben bookings Tabelle:

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

Wenn die Funktion einen Batch-DataFrame zurückgibt, @dp.table erzeugt stattdessen eine materialisierte Ansicht. Die einzige Ausnahme ist replace_where, was immer zu einer Streaming-Tabelle führt. Das folgende Beispiel hält den täglichen Umsatz für Check-ins am oder nach dem 1. Juli 2025 auf dem neuesten Stand, ohne frühere Termine neu zu berechnen:

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

Jede Ausführung löscht die Zeilen, die dem Prädikat entsprechen, und berechnet nur diesen Bereich neu. Siehe Batchverarbeitung mit REPLACE WHERE Flows.

Aktualisieren Sie eine Tabelle

Um eine Tabelle zu aktualisieren, die du mit einem Decorator definiert hast, führe den Code, der sie definiert, erneut aus, zum Beispiel indem du die Notebookzelle erneut ausführst, das ganze Notebook ausführst oder das Notebook als Job ausführst. Jeder Durchlauf erstellt die Tabelle, falls sie nicht existiert, und aktualisiert sie, falls doch.

Um alle in der Quelle verfügbaren Daten erneut zu verarbeiten, übergeben Sie full_refresh=True an einen der beiden 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")

Sie können keine REFRESH-Anweisung für eine mit einem Decorator definierte Tabelle verwenden oder Aktualisierungen mit SCHEDULE oder TRIGGER ON UPDATE planen. Um einen Zeitplan zu aktualisieren, definieren Sie die Tabelle in SQL oder planen Sie das Notizbuch als Job. Siehe Lakeflow Jobs.

Konfiguration der Tabelle

Die Dekoratoren akzeptieren dieselben gemeinsamen Datensatzparameter, die sie auch innerhalb einer Pipeline akzeptieren, einschließlich comment, table_properties, partition_cols, cluster_by, schema und 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 die Parameterliste siehe materialized_view und Tabelle.

private=True wird nicht unterstützt, da eine private Tabelle nur von anderen Datensätzen in derselben Pipeline gelesen werden kann.

Nicht unterstützte APIs

Eine eigenständige Tabelle ist ein einzelner Datensatz mit einem einzigen Fluss, daher sind die APIs, die Beziehungen zwischen Datensätzen beschreiben, nicht verfügbar. Die folgenden verursachen außerhalb einer Pipeline einen Fehler:

  • @dp.temporary_view und dp.create_streaming_table
  • @dp.append_flow und weitere zusätzliche Flows
  • dp.create_auto_cdc_flow und dp.create_auto_cdc_from_snapshot_flow
  • @dp.replace_flow und der Parameter replace_using, mit denen REPLACE USING-Abläufe definiert werden. Siehe Teilweises Ersetzen von Snapshots mit REPLACE USING-Flows.
  • dp.create_sink
  • Erwartungen, wie @dp.expect und @dp.expect_or_fail

Um diese zu verwenden, erstellen Sie stattdessen eine Lakeflow-Pipeline. Siehe Entwickeln von Pipelinecode mit Python.

Senden Sie SQL-Anweisungen mit spark.sql()

Übergeben Sie in einem Python-Notebook dieselben Anweisungen an spark.sql(), die Sie in einem Databricks SQL-Warehouse ausführen würden. Die Syntax für eigenständige materialisierte Ansichten und Streaming-Tabellen ist identisch; nur die Art, wie Sie die Anweisung übermitteln, unterscheidet sich. Wie bei einem Lager führt jede CREATE- oder REFRESH-Anweisung eine serverlose Pipeline aus, um den Vorgang zu verarbeiten.

Die spark Sitzung ist standardmäßig in Azure Databricks Notizbüchern verfügbar, sodass kein Import erforderlich ist.

Erstellen einer materialisierten Ansicht

Im folgenden Beispiel wird die materialisierte Sicht mv1 von der Basistabelle base_table1 erstellt:

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

Vollständige CREATE MATERIALIZED VIEW Details, z. B. geplante und ausgelöste Aktualisierungen, finden Sie unter Erstellen einer materialisierten Ansicht.

Erstellen Sie eine Streamingtabelle

Im folgenden Beispiel wird die Streamingtabelle sales aus der raw_data Tabelle erstellt:

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

Ausführliche CREATE STREAMING TABLE Informationen, einschließlich des Ladens von Dateien mit automatischem Laden und Planen, finden Sie unter Verwenden von eigenständigen Streamingtabellen.

Materialisierte Ansicht oder Streaming-Tabelle aktualisieren

Verwenden Sie eine REFRESH Anweisung, um eine eigenständige Tabelle mit den neuesten Daten aus der Quelle zu aktualisieren:

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

Bei serverlosen allgemeinen Berechnungen sind Aktualisierungen synchron. Asynchrone Aktualisierungen (das ASYNC Schlüsselwort) werden nicht unterstützt. Siehe Serverless allgemeine Rechenleistung.

Parametrisierte Anweisungen

Um Werte aus Ihrem Python-Code an eine Anweisung zu übergeben, anstatt sie fest zu codieren, verwenden Sie benannte Parametermarkierungen in der SQL-Anweisung und übergeben Sie deren Werte über das Argument args von spark.sql(). Verwenden Sie einen Platzhalter wie :min_sales direkt für Literalwerte. Umschließen Sie die Markierung IDENTIFIER() nur, wenn der Parameter ein Objektname ist, z. B. eine Tabelle, Ansicht oder ein Schema, da Bezeichner nicht als einfache Zeichenfolgenwerte ersetzt werden können.

Im folgenden Beispiel werden sowohl der materialisierte Ansichtsname als auch ein Filterwert parametrisiert:

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

Weitere Informationen finden Sie unter Parametermarkierungen und IDENTIFIER Klauseln.

Andere Anweisungen ausführen

Sie können jede eigenständige materialisierte Ansichts- oder Streamingtabellen-Anweisung aus einem Python-Notizbuch ausführen, indem Sie diese spark.sql()an übergeben, einschließlich Anweisungen zum Planen von Aktualisierungen, Ändern einer Tabelle oder Ablegen einer Tabelle. Informationen zur Verwendung materialisierter Ansichten und Streamingtabellen, einschließlich SQL-Syntax, finden Sie unter Verwenden von eigenständigen materialisierten Ansichten und Verwenden von eigenständigenStreamingtabellen.

Limitations

Eigenständige materialisierte Ansichten und Streamingtabellen, die auf serverlosen allgemeinen Berechnungen erstellt wurden, weisen zusätzliche Einschränkungen auf, z. B. keine Unterstützung für asynchrone Aktualisierungen und keine Kostenzuordnung pro Tabelle. Die vollständige Liste finden Sie unter "Serverless general compute".

Da diese Pipelines auf serverlosen allgemeinen Rechenressourcen und nicht in einem SQL-Warehouse ausgeführt werden, übernehmen sie keine benutzerdefinierten Tags von einem übergeordneten Warehouse. Die Weitergabe von Warehouse-Tags an system.billing.usage gilt nur für materialisierte Sichten und Streamingtabellen, deren Anweisungen von einem SQL-Warehouse ausgeführt werden. Siehe Attributkosten für das SQL-Lager mit benutzerdefinierten Tags.

Tabellen, die mit Pipeline-Dekoratoren definiert werden, unterliegen den folgenden zusätzlichen Einschränkungen:

  • Du kannst sie weder mit einer REFRESH Anweisung aktualisieren noch Aktualisierungen mit SCHEDULE oder TRIGGER ON UPDATE planen. Siehe Aktualisieren einer Tabelle.
  • Erwartungen, zusätzliche Flüsse, Change Data Capture (CDC)-Flüsse, Senken und temporäre Ansichten werden nicht unterstützt. Siehe Nicht unterstützte APIs.
  • private=True wird nicht unterstützt.

Weitere Ressourcen

  • Zuletzt aktualisiert um