Python skaler kullanıcı tanımlı fonksiyonlar (UDF'ler)

Python skaler UDF'ler, Azure Databricks'te SQL sorguları içinde özel Python mantığı çalıştırmanıza olanak tanır. Bu sayfa, Spark SQL'de bunları nasıl kaydedip çağıracağını, servis kimlik bilgilerini ve sırlarını nasıl kullanacağını, alt ifade değerlendirme sırası uyarılarını nasıl işleteceğimizi gösterir.

Gereksinimler

  • Databricks Runtime 12.2 LTS ve altında Python UDF'leri ve Pandas UDF'leri standart erişim modunu kullanan Unity Kataloğu işlemlerinde desteklenmez.

  • Skaler Python UDF'leri ve Pandas UDF'leri tüm erişim modları için Databricks Runtime 13.3 LTS ve üzerinde desteklenir.

  • Unity Catalog etkin kümelerde Python UDF'leri için ARM örneği desteği için Databricks Runtime 15.2 veya üzeri gerekir.

Databricks Runtime 14.0 ve altında, standart erişim modunu kullanan Unity Kataloğu kümelerinde Python UDF'leri ve Pandas UDF'leri desteklenmez. Skaler Python UDF'leri ve Pandas UDF'leri Databricks Runtime 14.1 ve üzeri tüm erişim modları için desteklenir.

Databricks Runtime 14.1 ve üzerinde SQL söz dizimlerini kullanarak skaler Python UDF'lerini Unity Kataloğu'na kaydedebilirsiniz. Bkz. Unity Kataloğu'nda SQL ve Python kullanıcı tanımlı işlevler (UDF' ler).

İşlevi UDF olarak kaydetme

def squared(s):
  return s * s
spark.udf.register("squaredWithPython", squared)

İsteğe bağlı olarak UDF'nizin dönüş türünü ayarlayabilirsiniz. Varsayılan dönüş türü şeklindedir StringType.

from pyspark.sql.types import LongType
def squared_typed(s):
  return s * s
spark.udf.register("squaredWithPython", squared_typed, LongType())

Spark SQL'de UDF'yi çağırma

spark.range(1, 20).createOrReplaceTempView("test")

%sql select id, squaredWithPython(id) as id_squared from test

DataFrame'lerle UDF kullanma

from pyspark.sql.functions import udf
from pyspark.sql.types import LongType
squared_udf = udf(squared, LongType())
df = spark.table("test")
display(df.select("id", squared_udf("id").alias("id_squared")))

Alternatif olarak, anotasyon söz dizimini kullanarak aynı UDF'yi bildirebilirsiniz.

from pyspark.sql.functions import udf

@udf("long")
def squared_udf(s):
  return s * s
df = spark.table("test")
display(df.select("id", squared_udf("id").alias("id_squared")))

UDF ile çeşitlemeler

Değişken için PySpark türü VariantType olarak belirtilir ve değerler VariantVal türündedir. Çeşitlemeler hakkında bilgi için bkz. Değişken verilerini sorgulama.

from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import VariantType, VariantVal

# Return Variant
@udf(returnType = VariantType())
def toVariant(jsonString):
  return VariantVal.parseJson(jsonString)

spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toVariant(col("json"))).display()

+---------------+
|toVariant(json)|
+---------------+
|        {"a":1}|
+---------------+

from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import StructField, StructType, VariantType, VariantVal

# Return Struct<Variant>
@udf(returnType = StructType([StructField("v", VariantType(), True)]))
def toStructVariant(jsonString):
  return {"v": VariantVal.parseJson(jsonString)}

spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toStructVariant(col("json"))).display()

+---------------------+
|toStructVariant(json)|
+---------------------+
|        {"v":{"a":1}}|
+---------------------+

from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import ArrayType, VariantType, VariantVal

# Return Array<Variant>
@udf(returnType = ArrayType(VariantType()))
def toArrayVariant(jsonString):
  return [VariantVal.parseJson(jsonString)]

spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toArrayVariant(col("json"))).display()

+--------------------+
|toArrayVariant(json)|
+--------------------+
|           [{"a":1}]|
+--------------------+

from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import MapType, StringType, VariantType, VariantVal

# Return Map<String, Variant>
@udf(returnType = MapType(StringType(), VariantType(), True))
def toMapVariant(jsonString):
  return {"v1": VariantVal.parseJson(jsonString), "v2": VariantVal.parseJson("[" + jsonString + "]")}

spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toMapVariant(col("json"))).display()

+-----------------------------+
|           toMapVariant(json)|
+-----------------------------+
|{"v2":[{"a":1}],"v1":{"a":1}}|
+-----------------------------+

UDF ile Dosyalar

Important

Bu özellik Beta sürümündedir.

Bir dosya için PySpark tipi FileType. Bunu bir UDF'de parametre veya dönüş tipi olarak, üst seviye tip olarak veya iç içe olarak kullanın. Tip, iç içleme kuralları ve FileRef API için FileType'a bakınız.

Bir dosyanın içeriğini bir UDF'de okumak için, açabileceğiniz yerel bir yol elde etmek üzere file.as_local_file() öğesini çağırın veya dosyanın baytlarını akış olarak okumak için file.open() öğesini çağırın. Python, Scala ve SQL'deki örnekler için, örneğin görüntü işleme, dosya tipi algılama ve video çerçeve çıkarma gibi konularda, bkz. UDF'lerle dosyaları işlemek. Tip referansı için bkz.FILE

Unity Kataloğu'nda (CREATE FUNCTION) kayıtlı bir UDF, bir FILEmeta veriyi okuyabilir ama içeriğini okuyamaz ve dosya oluşturamaz. Bir dosyanın içeriğini okumak için oturum kapsamlı bir UDF kullanın (open, as_local_file) veya bir dosya (from_bytes, ) from_local_fileoluşturun.

UDF gövdesinde her FILE değer bir FileRef nesnedir:

from pyspark.sql.functions import col, udf
from pyspark.sql.types import FileRef, StringType

@udf(returnType=StringType())
def file_content_type(file: FileRef) -> str:
  return file.content_type

df = spark.table("documents")
display(df.select(col("file").uri, file_content_type(col("file"))))

Değerlendirme sırası ve null denetimi

Spark SQL (SQL ve DataFrame ile Veri Kümesi API'si dahil) alt ifadelerin değerlendirilme sırasını garanti etmez. Özellikle, bir işlecin veya işlevin girişleri mutlaka soldan sağa veya başka bir sabit sırada değerlendirilmez. Örneğin, mantıksal AND ve OR ifadelerde soldan sağa "kısa devre" semantiği yoktur.

Bu nedenle, boole ifadelerinin yan etkilerine veya değerlendirme sırasına, ayrıca WHERE ve HAVING tümcelerinin dizilimine güvenmek tehlikelidir; çünkü bu ifadeler ve tümceler, sorgu iyileştirme ve planlama sırasında yeniden düzenlenebilir. Özellikle, bir UDF null denetim için SQL'de kısa devre semantiği kullanıyorsa, UDF'yi çağırmadan önce null denetimin gerçekleşeceğinin garantisi yoktur. Örneğin,

spark.udf.register("strlen", lambda s: len(s), "int")
spark.sql("select s from test1 where s is not null and strlen(s) > 1") # no guarantee

Bu WHERE madde, null değerleri filtreledikten sonra UDF'nin çağrılmasını garanti etmez strlen.

Doğru null denetimi gerçekleştirmek için aşağıdakilerden birini yapmanızı öneririz:

  • UDF'nin kendisini null algılayan hale getirin ve UDF'nin içinde null denetimi yapın
  • Null denetimi yapmak ve koşullu bir dalda UDF'yi çağırmak için IF veya CASE WHEN ifadelerini kullanın.
spark.udf.register("strlen_nullsafe", lambda s: len(s) if not s is None else -1, "int")
spark.sql("select s from test1 where s is not null and strlen_nullsafe(s) > 1") # ok
spark.sql("select s from test1 where if(s is not null, strlen(s), null) > 1")   # ok

Unity Kataloğu sırlarına erişim

Oturum kapsamlı Python UDF'den bir Unity Kataloğu sırrına erişmek için bkz. Oturum kapsamlı Python UDF'de bir sır kullanın. Skaler veya Batch Unity Catalog Python UDF'den bildirilen sırlara erişmek için bkz. Bir Python UDF'de sırlar kullan.

Python UDF'lerinde hizmet kimlik bilgileri

Oturum kapsamlı skaler Python UDF'ler ve skaler Unity Kataloğu Python UDF'leri, Unity Kataloğu hizmet kimlik bilgilerini kullanarak harici bulut servislerine güvenli bir şekilde erişebilir. Bu, bulut tabanlı belirteçleştirme, şifreleme veya gizli dizi yönetimi gibi işlemleri doğrudan veri dönüşümlerinizle tümleştirmek için kullanışlıdır.

Gereksinimler UDF türüne ve hesaplamaya göre değişiklik gösterir. Bkz. Python UDF'de bir hizmet kimlik bilgisi kullanma.

Hizmet kimlik bilgileri oluşturmak için bkz. Hizmet kimlik bilgileri oluşturma.

Oturum kapsamına bağlı skaler Python UDF'de bir servis kimlik bilgisi kullanın

Hizmet kimlik bilgilerine erişmek için UDF mantığınızdaki yardımcı programını kullanarak databricks.service_credentials.getServiceCredentialsProvider() bulut SDK'larını uygun kimlik bilgileriyle başlatın. Tüm kod UDF gövdesinde kapsüllenmelidir.

@udf
def use_service_credential():
    from azure.mgmt.web import WebSiteManagementClient

    # Assuming there is a service credential named 'testcred' set up in Unity Catalog
    web_client = WebSiteManagementClient(subscription_id, credential = getServiceCredentialsProvider('testcred'))
    # Use web_client to perform operations

Hizmet kimlik bilgileri izinleri

Oturum kapsamlı UDF'ler, arayanın izinlerini kullanır. Gerekli ayrıcalıklar için bkz. Bir Python UDF'de hizmet kimlik bilgisi kullanın.

Oturum kapsamındaki UDF'ler için hesaplama düzeyinde varsayılan kimlik bilgileri

Skaler Python UDF'lerde kullanıldığında, Databricks otomatik olarak hesaplama ortamı değişkeninden varsayılan hizmet kimlik bilgilerini kullanır. Bu davranış, UDF kodunuzda kimlik bilgileri takma adlarını açıkça yönetmeden dış hizmetlere güvenli bir şekilde başvurmanızı sağlar. Bkz . İşlem kaynağı için varsayılan hizmet kimlik bilgilerini belirtme

Varsayılan kimlik bilgisi desteği yalnızca Standart ve Ayrılmış erişim modu kümelerinde kullanılabilir. SQL depolarında mevcut değildir.

Sağlayıcıyı kullanmak için azure-identity paketini yüklemeniz gerekir. Paketi yüklemek için bkz Defter kapsamlı Python kütüphaneleri veya Hesaplama kapsamlı kütüphaneler.

@udf
def use_service_credential():
    from azure.identity import DefaultAzureCredential
    from azure.mgmt.web import WebSiteManagementClient

    # DefaultAzureCredential is automatically using the default service credential for the compute
    web_client_default = WebSiteManagementClient(DefaultAzureCredential(), subscription_id)

    # Use web_client to perform operations

Skaler Unity Kataloğu Python UDF'de bir hizmet kimlik bilgisi kullanın

Hizmet kimlik bilgilerini UDF tanımındaki CREDENTIALS yan tümcesinde belirtin. Bir kimlik bilgisini DEFAULT işaretleyerek yamalı bulut SDK'larının otomatik olarak kullanmasını sağlayabilirsiniz. Klasik hesaplamada, bu özellik Databricks Runtime 18.1 veya üzeri gerektirir. Sunucusuz işlemde ve Pro ile sunucusuz SQL ambarlarında, UDF'nin environment_version değerini açıkça 6 veya üzeri olarak ayarlayın. Tam hesaplama, ağ ve izin gereksinimleri için bkz. Use a service credential in a Python UDF.

Görev yürütme bağlamını al

Kullanıcının kimliği, küme etiketleri, spark görev kimliği ve daha fazlası gibi bağlam bilgilerini almak için TaskContext PySpark API'sini kullanın. Bkz. UDF'de görev bağlamı alma.

Sınırlamalar

PySpark UDF'leri için aşağıdaki sınırlamalar geçerlidir:

  • Dosya erişim kısıtlamaları: Databricks Runtime 14.2 ve altında, paylaşılan kümelerdeki PySpark UDF'leri Git klasörlerine, çalışma alanı dosyalarına veya Unity Katalog Birimlerine erişemez.

  • Yayın değişkenleri: Standart erişim modu kümelerindeki PySpark UDF'leri ve sunucusuz işlem yayın değişkenlerini desteklemez.

  • Sunucusuz üzerinde bellek sınırı: Sunucusuz işlemdeki PySpark UDF'leri, PySpark UDF başına 1 GB bellek sınırına sahiptir. Bu sınırın aşılması, UDF_PYSPARK_USER_CODE_ERROR türünde bir hatayla sonuçlanır . MEMORY_LIMIT_SERVERLESS.
  • Standart erişim modunda bellek sınırı: Standart erişim modundaki PySpark UDF'leri, seçilen örnek türünün kullanılabilir belleğine göre bir bellek sınırına sahiptir. Kullanılabilir belleğin aşılması, UDF_PYSPARK_USER_CODE_ERROR türünde bir hatayla sonuçlanır . MEMORY_LIMIT.
  • Sunucusuz SQL ambarlarında ağ erişimi: Varsayılan olarak, sunucusuz SQL ambarlarındaki Python UDF'ler giden ağ istekleri yapamaz ve ağ çağrıları yapmaya çalışan sorgular süresiz olarak kilitlenebilir. Giden ağ erişimini etkinleştirmek için, Çalışma alanınızın Önizlemeler sayfasındakiSunucusuz SQL Ambarları'nda yalıtılmış iş yükleri için ağı etkinleştirme Genel Önizleme özelliğini etkinleştirin. Aksi takdirde, ağ erişimi gerektiren UDF'ler için sunucusuz işlem veya klasik işlem kullanın.
  • Son güncelleştirme: