Tek başına işlem hatlarıyla Python kullanma

Important

Sunucusuz genel işlem üzerinde bir not defteri üzerinden bağımsız somutlaştırılmış görünümler ve akış tabloları oluşturma ve yenileme Beta aşamasındadır ve yalnızca belirli bölgelerde kullanılabilir. Bkz. Not Defterleri.

Python kullanarak bir defterden bağımsız maddi görünümler ve akış tabloları oluşturabilir ve yenileyebilirsiniz. Bu, tek başına işlem hatlarını diğer Python tabanlı not defteri iş akışlarınızla birlikte yönetmenizi sağlar.

Bunu yapmanın iki yolu vardır:

  • Tabloyu @dp.materialized_view, pyspark.pipelines ve @dp.table dekoratörleriyle tanımlayın. Mantığın DataFrame kodu olarak ifade edilmesi daha kolay olduğunda bunu kullanın. Bkz. Boru hattı dekoratörlerinin bulunduğu tabloları tanımla.
  • Bir Databricks SQL ambarının çalıştırdığı SQL ifadelerinin aynısını, bunları spark.sql() öğesine ileterek gönderin. Bu, size materyalize görünüm ve akış tablosu için REFRESH deyimleri ve yenileme zamanlamaları dahil tam, bağımsız SQL kapsamını sunar. spark.sql() ile SQL ifadeleri gönderme konusuna bakın.

Tek başına işlem hatları için Python kaynağı, sunucusuz genel işlem için bir not defteri eklenmesini gerektirir. Databricks SQL veri ambarından Python ile bağımsız işlem hatları oluşturamaz veya yenileyemezsiniz; çünkü veri ambarı Python not defterlerini değil, SQL ifadelerini çalıştırır. Bunun yerine bir SQL veri ambarı kullanmak için bkz. Bağımsız somutlaştırılmış görünümleri kullanma ve Bağımsız akış tablolarını kullanma.

Requirements

Python ile bağımsız işlem hatları oluşturmak ve yenilemek için, Databricks Runtime 18.1 veya üzerini çalıştıran sunucusuz genel işlem kaynağına bağlı bir not defterine ihtiyacınız vardır. Bölgesel kullanılabilirlik ve izinler de dahil olmak üzere gereksinimlerin tam listesi için bkz. Not defterleri.

İşlem hattı dekoratörleriyle tablolar tanımlayın

Lakeflow pipeline'da kullandığınız dekoratörlerle bağımsız bir maddeleştirilmiş görünüm veya akış tablosu tanımlayabilirsiniz. Her süslü fonksiyon bir masa tanımlar. Hücreyi çalıştırdığınızda, Azure Databricks tabloyu oluşturur ve onu doldurmak için sunucusuz bir boru hattı çalıştırır. Güncelleme bittiğinde hücre geri döner.

Warning

İşlem hattı dekoratörleri sunucusuz ortam sürümü 5 veya üzerini gerektirir.

Maddeleştirilmiş bir görünüm tanımlayın

Batch DataFrame döndüren bir fonksiyonda @dp.materialized_view kullanın. Aşağıdaki örnek, Wanderbricks örnek veri setindeki tablodan maddeleştirilmiş görünümü daily_booking_revenuebookings oluşturur:

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

Akış okumasından bir tablo tanımlamak için @dp.table yerine ' kullanın.

Bir akış tablosu tanımlayın

Akışlı DataFrame döndüren bir fonksiyonda @dp.table kullanın. Aşağıdaki örnek, aynı bookings_raw tablosunun akışlı okunmasından bookings akış tablosunu oluşturur:

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

Fonksiyon toplu bir DataFrame dönerse, @dp.table bunun yerine maddeleştirilmiş bir görünüm oluşturur. Tek istisna, her zaman bir akış tablosu sonucunu veren replace_where'dır. Aşağıdaki örnek, 1 Temmuz 2025 ve sonrasındaki check-in’ler için günlük geliri, önceki tarihleri yeniden hesaplamadan güncel tutar:

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

Her çalıştırma, koşulla eşleşen satırları siler ve yalnızca o aralığı yeniden hesaplar. Bkz. REPLACE WHERE akışlarıyla toplu işleme.

Bir masayı yenile

Bir dekoratör ile tanımladığınız bir tabloyu yenilemek için, defter hücresini yeniden çalıştırarak, tüm defteri çalıştırarak veya not defterini iş olarak çalıştırarak onu tanımlayan kodu tekrar çalıştırın. Her çalıştırma, tablo mevcut değilse onu oluşturur; mevcutsa yeniler.

Kaynakta bulunan tüm verileri yeniden işlemek için, iki dekoratörden herhangi birine full_refresh=True parametresini iletin:

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

Dekoratörle tanımlanmış bir tabloda REFRESH deyimini kullanamazsınız ya da yenilemeleri SCHEDULE veya TRIGGER ON UPDATE ile zamanlayamazsınız. Zamanlanmış olarak yenilemek için tabloyu SQL'de tanımlayın veya not defterini görev olarak zamanlayın. Bakınız Lakeflow İşleri.

Tabloyu yapılandırma

Dekoratörler, işlem hattında da kabul edilen aynı ortak veri kümesi parametrelerini kabul eder; bunlara comment, table_properties, partition_cols, cluster_by, schema ve spark_conf dâhildir:

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

Parametre listesi için materialized_view ve tabloya bakınız.

private=True desteklenmez, çünkü özel bir tablo yalnızca aynı boru hattındaki diğer veri setleri tarafından okunabilir.

Desteklenmeyen API'ler

Bağımsız bir tablo, tek bir akışa sahip tek bir veri setisidir, bu yüzden veri kümeleri arasındaki ilişkileri tanımlayan API'ler mevcut değildir. Aşağıdakiler, boru hattı dışında hataya neden olur:

  • @dp.temporary_view ve dp.create_streaming_table
  • @dp.append_flow ve diğer ek akışlar
  • dp.create_auto_cdc_flow ve dp.create_auto_cdc_from_snapshot_flow
  • REPLACE USING akışlarını tanımlayan @dp.replace_flow ve replace_using parametresi. Bkz. REPLACE USING akışlarıyla kısmi anlık görüntü değiştirme.
  • dp.create_sink
  • Beklentiler, örneğin @dp.expect ve @dp.expect_or_fail

Bunları kullanmak için bunun yerine bir Lakeflow işlem hattı oluşturun. Bkz. Pythonile işlem hattı kodu geliştirme.

spark.sql() ile SQL deyimlerini gönder

Bir Python not defterinde, Databricks SQL ambarından çalıştıracağınız ifadelerin aynısını spark.sql() öğesine iletin. Bağımsız maddileştirilmiş görünüm ve akış tablosu söz dizimi aynıdır; yalnızca ifadeyi gönderme biçiminiz farklıdır. Bir ambarda olduğu gibi, her CREATE veya REFRESH deyimi işlemi işlemek için sunucusuz bir işlem hattı çalıştırır.

Oturum spark varsayılan olarak Azure Databricks not defterlerinde kullanılabilir, bu nedenle içeri aktarma gerekmez.

Gerçekleştirilmiş görünüm oluşturma

Aşağıdaki örnek, temel tablodan mv1gerçekleştirilmiş görünümü base_table1 oluşturur:

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

Zamanlanmış ve tetiklenen yenilemeler gibi tüm CREATE MATERIALIZED VIEW ayrıntılar için bkz. Gerçekleştirilmiş görünüm oluşturma.

Akış tablosu oluşturma

Aşağıdaki örnek, akış tablosunu tablodan salesraw_data oluşturur:

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

Otomatik Yükleyici ile dosya yükleme ve zamanlama gibi tüm CREATE STREAMING TABLE ayrıntılar için bkz. Tek başına akış tablolarını kullanma.

Somutlaştırılmış görünümü veya akış tablosunu yenile

Bağımsız bir tabloyu kaynağındaki en güncel verilerle güncellemek için REFRESH deyimini kullanın:

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

Sunucusuz genel işlemde yenilemeler zaman uyumludur. Zaman uyumsuz yenilemeler ( ASYNC anahtar sözcük) desteklenmez. Bkz . Sunucusuz genel işlem.

Deyimleri parametreleştirme

Python kodunuzdaki değerleri sabit olarak kodlamak yerine bir SQL ifadesine aktarmak için, SQL'de adlandırılmış parametre işaretçileri kullanın ve bu değerleri args öğesinin spark.sql() bağımsız değişkeni aracılığıyla sağlayın. Değişmez değerler için doğrudan :min_sales gibi bir işaretleyici kullanın. İşaretçiyi IDENTIFIER() yalnızca parametre tablo, görünüm veya şema gibi bir nesne adı olduğunda kaydırır, çünkü tanımlayıcılar düz dize değerleri olarak değiştirilemez.

Aşağıdaki örnek hem gerçekleştirilmiş görünüm adını hem de filtre değerini parametreleştirir:

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

Daha fazla bilgi için bkz.: Parametre işaretçileri ve IDENTIFIER yan tümcesi.

Diğer ifadeleri çalıştırma

Python not defterinden, herhangi bir bağımsız somutlaştırılmış görünüm veya akış tablosu deyimini spark.sql()'a ileterek çalıştırabilirsiniz; buna yenilemeyi zamanlama, bir tabloyu değiştirme veya bir tabloyu silme deyimleri de dahildir. SQL söz dizimi dahil olmak üzere gerçekleştirilmiş görünümlerin ve akış tablolarının nasıl kullanılacağını anlamak için bkz. Tek başına gerçekleştirilmiş görünümleri kullanma ve Tek başına akış tablolarını kullanma.

Limitations

Sunucusuz genel işlemde oluşturulan tek başına gerçekleştirilmiş görünümler ve akış tabloları, zaman uyumsuz yenilemeler için destek olmaması ve tablo başına maliyet atfı olmaması gibi ek sınırlamalara sahiptir. Tam liste için bkz. Sunucusuz genel işlem.

Bu boru hatları SQL deposu yerine sunucusuz genel hesaplama üzerinde çalıştıkları için, kapalı bir depodan özel etiketler devralmazlar. system.billing.usage için ambar etiketi aktarımı yalnızca sorguları bir SQL ambarında çalışan somutlaştırılmış görünümler ve akış tabloları için geçerlidir. Bkz. Özel etiketlerle maliyetleri SQL veri ambarına atfetme.

Boru hattı dekoratörleriyle tanımlanan tablolar aşağıdaki ek sınırlamalara sahiptir:

  • Bunları bir REFRESH açıklama ile yenileyemezsin veya güncellemeleri planlayamazsınSCHEDULE.TRIGGER ON UPDATE Bkz. Tabloyu yenile.
  • Beklentiler, ek akışlar, değişiklik veri yakalama (CDC) akışları, yutucular ve geçici görünümler desteklenmemektedir. Desteklenmeyen API'lere bakınız.
  • private=True desteklenmez.

Ek kaynaklar

  • Son güncelleştirme: