Menggunakan Python dengan alur mandiri

Important

Membuat dan menyegarkan tampilan terwujud mandiri dan tabel streaming dari buku catatan pada komputasi umum tanpa server ada di Beta dan tersedia di wilayah tertentu. Lihat Notebook.

Anda dapat membuat dan memuat ulang tampilan materialisasi mandiri serta tabel streaming dari notebook menggunakan Python. Ini memungkinkan Anda mengelola alur mandiri bersama alur kerja notebook berbasis Python lainnya.

Ada dua cara untuk melakukan ini:

• Definisikan tabel dengan dekorator pyspark.pipelines, @dp.materialized_view, dan @dp.table. Gunakan ini ketika logika lebih mudah diekspresikan sebagai kode DataFrame. Lihat Mendefinisikan tabel dengan dekorator pipeline.
• Kirim pernyataan SQL yang sama yang dijalankan oleh databricks SQL warehouse dengan meneruskannya ke spark.sql(). Ini memberi Anda seluruh cakupan SQL untuk materialized view mandiri dan tabel streaming, termasuk pernyataan REFRESH dan jadwal penyegaran. Lihat Kirim pernyataan SQL dengan spark.sql().

Python sumber untuk alur mandiri memerlukan notebook yang dilampirkan ke komputasi umum tanpa server. Anda tidak dapat menggunakan Python untuk membuat atau menyegarkan pipeline mandiri dari gudang SQL Databricks, karena gudang SQL hanya menjalankan pernyataan SQL, bukan notebook Python. Untuk menggunakan gudang SQL sebagai gantinya, lihat Menggunakan tampilan terwujud mandiri dan Menggunakan tabel streaming mandiri.

Requirements

Untuk membuat dan memperbarui pipeline mandiri dengan Python, Anda memerlukan notebook yang terhubung ke komputasi umum nirserver pada Databricks Runtime 18.1 atau yang lebih baru. Untuk daftar lengkap persyaratan, termasuk ketersediaan dan izin regional, lihat Notebook.

Definisikan tabel dengan dekorator alur

Anda dapat mendefinisikan tampilan materialisasi mandiri atau tabel streaming dengan dekorator yang sama seperti yang Anda gunakan dalam pipeline Lakeflow. Setiap fungsi yang diberi dekorator mendefinisikan satu tabel. Saat Anda menjalankan sel, Azure Databricks membuat tabel dan menjalankan pipeline serverless untuk mengisinya. Sel akan muncul kembali setelah pembaruan selesai.

Warning

Dekorator pipeline memerlukan versi lingkungan serverless 5 atau yang lebih baru.

Definisikan tampilan materialisasi

Gunakan @dp.materialized_view pada fungsi yang mengembalikan batch DataFrame. Contoh berikut membuat tampilan terwujud daily_booking_revenue dari tabel bookings di set data contoh 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"))
  )

Untuk mendefinisikan tabel dari bacaan streaming, gunakan @dp.table sebagai gantinya.

Definisikan tabel streaming

Gunakan @dp.table pada fungsi yang mengembalikan DataFrame streaming. Contoh berikut membuat tabel streaming bookings_raw dari pembacaan streaming pada tabel bookings yang sama:

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

Jika fungsi mengembalikan batch DataFrame, @dp.table maka akan membuat tampilan materialisasi sebagai gantinya. Satu pengecualian adalah replace_where, yang selalu menghasilkan tabel streaming. Contoh berikut menjaga pendapatan harian untuk check-in pada atau setelah 1 Juli 2025 tetap terbaru, tanpa menghitung ulang tanggal sebelumnya:

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

Setiap run menghapus baris yang sesuai dengan predikat dan hanya menghitung ulang rentang tersebut. Lihat Pemrosesan batch dengan alur REPLACE WHERE.

Muat ulang tabel

Untuk memuat ulang tabel yang Anda definisikan dengan dekorator, jalankan kembali kode yang mendefinisikan tabel tersebut, misalnya dengan menjalankan ulang sel notebook, menjalankan notebook secara keseluruhan, atau menjalankan notebook sebagai job. Setiap eksekusi membuat tabel jika belum ada dan memperbaruinya jika sudah ada.

Untuk memproses ulang semua data yang tersedia di sumber, teruskan full_refresh=True ke salah satu dekorator:

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

Anda tidak dapat menggunakan statemen REFRESH terhadap tabel yang didefinisikan dengan dekorator, atau menjadwalkan penyegaran dengan SCHEDULE atau TRIGGER ON UPDATE. Agar dapat diperbarui secara terjadwal, definisikan tabel di SQL, atau jadwalkan notebook sebagai tugas. Lihat Pekerjaan Lakeflow.

Konfigurasi tabel

Para dekorator menerima parameter dataset umum yang sama seperti yang mereka terima di dalam pipeline, termasuk comment, table_properties, partition_cols, cluster_by, schema, dan 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"))
  )

Untuk daftar parameter, lihat materialized_view dan tabel.

private=True tidak didukung, karena tabel privat hanya dapat dibaca oleh dataset lain dalam pipeline yang sama.

API yang tidak didukung

Tabel mandiri adalah satu dataset dengan satu alur, sehingga API yang mendeskripsikan hubungan antar dataset tidak tersedia. Berikut ini menimbulkan error di luar pipeline:

• @dp.temporary_view dan dp.create_streaming_table
• @dp.append_flow dan aliran tambahan lainnya
• dp.create_auto_cdc_flow dan dp.create_auto_cdc_from_snapshot_flow
• @dp.replace_flow dan parameter replace_using, yang menentukan alur REPLACE USING. Lihat Penggantian snapshot parsial dengan flow REPLACE USING.
• dp.create_sink
• Ekspektasi, seperti @dp.expect dan @dp.expect_or_fail

Untuk menggunakan ini, buatlah pipeline Lakeflow sebagai gantinya. Lihat Mengembangkan kode alur dengan Python.

Kirim pernyataan SQL dengan spark.sql()

Dalam buku catatan Python, berikan pernyataan yang sama yang akan Anda jalankan dari gudang Databricks SQL ke spark.sql(). Tampilan terwujud mandiri dan sintaks tabel streaming identik; hanya cara Anda mengirimkan pernyataan yang berbeda. Seperti pada gudang data, setiap pernyataan CREATE atau REFRESH menjalankan pipeline nirserver untuk memproses operasi.

Sesi spark tersedia secara default di buku catatan Azure Databricks, sehingga tidak diperlukan impor.

Buat tampilan terwujud

Contoh berikut membuat tampilan mv1 materialisasi dari tabel base_table1dasar :

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

Untuk detail lengkap CREATE MATERIALIZED VIEW, seperti refresh terjadwal dan refresh yang dipicu, lihat Membuat tampilan termaterialisasi.

Buat tabel streaming

Contoh berikut ini membuat tabel streaming sales dari tabel raw_data:

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

Untuk detail selengkapnya CREATE STREAMING TABLE , termasuk memuat file dengan Auto Loader dan penjadwalan, lihat Menggunakan tabel streaming mandiri.

Muat ulang tampilan terwujud atau tabel streaming

REFRESH Gunakan pernyataan untuk memperbarui tabel mandiri dengan data terbaru dari sumbernya:

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

Pada komputasi umum nirserver, penyegaran dilakukan secara sinkron. Penyegaran asinkron (kata kunci ASYNC) tidak didukung. Lihat Komputasi umum tanpa server.

Jadikan pernyataan berparameter

Untuk meneruskan nilai dari kode Python Anda ke dalam pernyataan alih-alih mengodekannya secara permanen, gunakan penanda parameter bernama di SQL dan berikan nilainya melalui args argumen spark.sql(). Gunakan penanda seperti :min_sales secara langsung untuk nilai harfiah. Bungkus penanda IDENTIFIER() hanya jika parameter adalah nama objek, seperti tabel, tampilan, atau skema, karena pengidentifikasi tidak dapat diganti sebagai nilai string biasa.

Contoh berikut membuat parameter nama tampilan materialisasi dan nilai filter:

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

Untuk informasi selengkapnya, lihat penanda parameter dan klausa IDENTIFIER.

Jalankan pernyataan lainnya

Anda dapat menjalankan tampilan terwujud mandiri atau pernyataan tabel streaming dari buku catatan Python dengan meneruskannya ke spark.sql(), termasuk pernyataan untuk menjadwalkan refresh, mengubah tabel, atau menghilangkan tabel. Untuk memahami cara menggunakan tampilan materialisasi dan tabel streaming, termasuk sintaks SQL, lihat Menggunakan tampilan materialisasi mandiri dan Menggunakan tabel streaming mandiri.

Keterbatasan

Tampilan terwujud mandiri dan tabel streaming yang dibuat pada komputasi umum tanpa server memiliki batasan tambahan, seperti tidak ada dukungan untuk refresh asinkron dan tidak ada atribusi biaya per tabel. Untuk daftar lengkapnya, lihat Komputasi umum tanpa server.

Karena pipeline ini berjalan pada komputasi umum nirserver, bukan pada warehouse SQL, pipeline ini tidak mewarisi tag kustom dari warehouse induk. Propagasi tag gudang ke system.billing.usage hanya berlaku untuk tampilan materialisasi dan tabel streaming yang pernyataannya dijalankan dari gudang SQL. Lihat Biaya atribut ke gudang SQL dengan tag kustom.

Tabel yang didefinisikan dengan dekorator pipeline memiliki keterbatasan tambahan berikut:

• Anda tidak bisa memperbaruinya dengan REFRESH pernyataan atau menjadwalkan penyegaran dengan SCHEDULE atau TRIGGER ON UPDATE. Lihat Segarkan tabel.
• Ekspektasi, aliran tambahan, aliran penangkapan data perubahan (CDC), sink, dan tampilan sementara tidak didukung. Lihat API yang Tidak Didukung.
• private=True tidak didukung.

Sumber daya tambahan

• Persyaratan untuk alur mandiri
• Gunakan tampilan materialisasi mandiri
• Menggunakan tabel streaming mandiri
• Jadwalkan penyegaran
• Refresh penuh untuk tabel streaming
• Referensi bahasa Python untuk pipeline Lakeflow