Not
Bu sayfaya erişim yetkilendirme gerektiriyor. Oturum açmayı veya dizinleri değiştirmeyi deneyebilirsiniz.
Bu sayfaya erişim yetkilendirme gerektiriyor. Dizinleri değiştirmeyi deneyebilirsiniz.
Pandas kullanıcı tanımlı bir işlev (UDF) (vektörleştirilmiş UDF olarak da bilinir), verileri aktarmak için Apache Arrow ve verilerle çalışmak için pandas kullanan bir işlevdir. pandas UDF'leri, python UDF'lerine kıyasla performansı 100 kata kadar artırabilen vektörleştirilmiş işlemlere izin verir.
Arka plan bilgileri için Apache Spark 3.0'ın Yaklaşan SürümündeKi Yeni Pandas UDF'leri ve Python Tür İpuçları blog gönderisine bakın.
Anahtar sözcüğünü pandas_udf bir dekoratör olarak kullanarak bir pandas UDF tanımlarsınız ve işlevi bir Python tür ipucu ile sarmalarsınız.
Bu makalede pandas UDF'lerinin farklı türleri açıklanır ve tür ipuçlarıyla pandas UDF'lerinin nasıl kullanılacağı gösterilir.
Seriden Seriye UDF
Skaler işlemleri vektörleştirmek için Series to Series pandas UDF kullanırsınız.
Bunları select ve withColumn gibi API'lerle kullanabilirsiniz.
Python işlevinin bir pandas Serisini giriş olarak kabul etmesi ve aynı uzunlukta bir pandas Serisi döndürmesi gerekir. Python tür ipuçlarını kullanarak bu türleri belirtin. Spark, verileri satır gruplarına bölerek, her toplu işlem için işlevi çağırarak ve ardından sonuçları birleştirerek bir pandas UDF çalıştırır.
Aşağıdaki örnekte, 2 sütunun çarpımını hesaplayan bir pandas UDF'nin nasıl oluşturulacağı gösterilmektedir.
import pandas as pd
from pyspark.sql.functions import col, pandas_udf
from pyspark.sql.types import LongType
# Declare the function and create the UDF
def multiply_func(a: pd.Series, b: pd.Series) -> pd.Series:
return a * b
multiply = pandas_udf(multiply_func, returnType=LongType())
# The function for a pandas_udf should be able to execute with local pandas data
x = pd.Series([1, 2, 3])
print(multiply_func(x, x))
# 0 1
# 1 4
# 2 9
# dtype: int64
# Create a Spark DataFrame, 'spark' is an existing SparkSession
df = spark.createDataFrame(pd.DataFrame(x, columns=["x"]))
# Execute function as a Spark vectorized UDF
df.select(multiply(col("x"), col("x"))).show()
# +-------------------+
# |multiply_func(x, x)|
# +-------------------+
# | 1|
# | 4|
# | 9|
# +-------------------+
Seriyi Yineleyiciden Seri UDF Yineleyicisine
Yineleyici UDF, aşağıdakiler dışında skaler pandas UDF ile aynıdır:
- Python işlevi
- Giriş olarak tek bir giriş toplu işlemi yerine toplu iş yineleyicisini alır.
- Tek bir çıkış toplu işlemi yerine çıkış toplu işlemlerinin yineleyicisini döndürür.
- Yineleyicideki çıkışın tamamının uzunluğu, girişin tamamının uzunluğuyla aynı olmalıdır.
- Sarmalanan pandas UDF, giriş olarak tek bir Spark sütunu alır.
Python tür ipucunu Iterator[pandas.Series] ->Iterator[pandas.Series] olarak belirtmelisiniz.
Bu pandas UDF, UDF yürütmesi bazı durumların başlatılmasını gerektirdiğinde (örneğin, her giriş toplu işlemine çıkarım uygulamak için bir makine öğrenmesi modeli dosyası yükleme) yararlı olur.
Aşağıdaki örnekte yineleyici desteğiyle pandas UDF'nin nasıl oluşturulacağı gösterilmektedir.
import pandas as pd
from typing import Iterator
from pyspark.sql.functions import col, pandas_udf, struct
pdf = pd.DataFrame([1, 2, 3], columns=["x"])
df = spark.createDataFrame(pdf)
# When the UDF is called with the column,
# the input to the underlying function is an iterator of pd.Series.
@pandas_udf("long")
def plus_one(batch_iter: Iterator[pd.Series]) -> Iterator[pd.Series]:
for x in batch_iter:
yield x + 1
df.select(plus_one(col("x"))).show()
# +-----------+
# |plus_one(x)|
# +-----------+
# | 2|
# | 3|
# | 4|
# +-----------+
# In the UDF, you can initialize some state before processing batches.
# Wrap your code with try/finally or use context managers to ensure
# the release of resources at the end.
y = 1 # value captured by the UDF closure
@pandas_udf("long")
def plus_y(batch_iter: Iterator[pd.Series]) -> Iterator[pd.Series]:
try:
for x in batch_iter:
yield x + y
finally:
pass # release resources here, if any
df.select(plus_y(col("x"))).show()
# +---------+
# |plus_y(x)|
# +---------+
# | 2|
# | 3|
# | 4|
# +---------+
Birden çok Seriyi Yineleyiciden Seri UDF Yineleyicisine
Birden çok Serinin Yineleyicisinden Seri UDF'nin Yineleyicisine Yineleyici, Serinin Yineleyicisi ile Seri UDF'nin Yineleyicisi arasında benzer özelliklere ve kısıtlamalara sahiptir. Belirtilen işlev toplu işlemlerin yineleyicisini alır ve bir toplu iş yineleyicisi çıkarır. UDF yürütmesi bazı durumların başlatılmasını gerektirdiğinde de yararlıdır.
Farklar şunlardır:
- Temel alınan Python işlevi pandas Series demetinin yineleyicisini alır.
- Sarmalanan pandas UDF, giriş olarak birden fazla Spark sütununu alır.
Iterator[Tuple[pandas.Series, ...]] tür ipuçlarını >-Iterator[pandas.Series] olarak belirtirsiniz.
from typing import Iterator, Tuple
import pandas as pd
from pyspark.sql.functions import col, pandas_udf, struct
pdf = pd.DataFrame([1, 2, 3], columns=["x"])
df = spark.createDataFrame(pdf)
@pandas_udf("long")
def multiply_two_cols(
iterator: Iterator[Tuple[pd.Series, pd.Series]]) -> Iterator[pd.Series]:
for a, b in iterator:
yield a * b
df.select(multiply_two_cols("x", "x")).show()
# +-----------------------+
# |multiply_two_cols(x, x)|
# +-----------------------+
# | 1|
# | 4|
# | 9|
# +-----------------------+
Seriden skaler Kullanıcı Tanımlı Fonksiyona (UDF)
Skaler pandas UDF'lerine seriler Spark toplama işlevlerine benzer.
A Series to scalar pandas UDF, bir veya daha fazla pandas Serisinden bir skaler değere bir toplama tanımlar ve burada her pandas Serisi bir Spark sütununu temsil eder.
tr-TR: Bir Seri'yi skaler pandas UDF için select, withColumn, groupBy.agg ve pyspark.sql.Window gibi API'lerle kullanırsınız.
Tür ipucunu pandas.Series, ... -> olarak Any ifade edebilirsiniz. Dönüş türü ilkel bir veri türü olmalı ve döndürülen skaler bir Python ilkel türü olabilir, örneğin int veya float veya numpy.int64gibi numpy.float64 bir NumPy veri türü olabilir.
Any ideal olarak belirli bir skaler tür olmalıdır.
Bu UDF türü kısmi toplamayı desteklemez ve her grubun tüm verileri belleğe yüklenir.
Aşağıdaki örnekte, , selectve groupBy işlemleriyle windowortalamayı hesaplamak için bu UDF türünün nasıl kullanılacağı gösterilmektedir:
import pandas as pd
from pyspark.sql.functions import pandas_udf
from pyspark.sql import Window
df = spark.createDataFrame(
[(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)],
("id", "v"))
# Declare the function and create the UDF
@pandas_udf("double")
def mean_udf(v: pd.Series) -> float:
return v.mean()
df.select(mean_udf(df['v'])).show()
# +-----------+
# |mean_udf(v)|
# +-----------+
# | 4.2|
# +-----------+
df.groupby("id").agg(mean_udf(df['v'])).show()
# +---+-----------+
# | id|mean_udf(v)|
# +---+-----------+
# | 1| 1.5|
# | 2| 6.0|
# +---+-----------+
w = Window \
.partitionBy('id') \
.rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
df.withColumn('mean_v', mean_udf(df['v']).over(w)).show()
# +---+----+------+
# | id| v|mean_v|
# +---+----+------+
# | 1| 1.0| 1.5|
# | 1| 2.0| 1.5|
# | 2| 3.0| 6.0|
# | 2| 5.0| 6.0|
# | 2|10.0| 6.0|
# +---+----+------+
Ayrıntılı kullanım için bkz . pyspark.sql.functions.pandas_udf.
Kullanım
Arrow toplu boyutunu ayarlama
Not
Bu yapılandırmanın sunucusuz işlem veya standart erişim modu ve Databricks Runtime 13.3 LTS ile 14.2 arasında yapılandırılmış işlem üzerinde hiçbir etkisi yoktur. Sunucusuz bilgi işlemde platform, Arrow toplu iş boyutlandırmasını dahili olarak yönetir.
Spark'taki veri bölümleri, JVM'de geçici olarak yüksek bellek kullanımına neden olabilecek Arrow kayıt gruplarına dönüştürülür. Bellek dışı özel durumları önlemek için, spark.sql.execution.arrow.maxRecordsPerBatch yapılandırmasını her toplu iş için en fazla satır sayısını belirleyen bir tamsayıya ayarlayarak Ok kaydı toplu işlemlerinin boyutunu ayarlayabilirsiniz. Varsayılan değer, toplu iş başına 10.000 kayıttır. Sütun sayısı büyükse, değer buna göre ayarlanmalıdır. Bu sınır kullanıldığında, her veri bölümü işlenmek üzere 1 veya daha fazla kayıt toplu işlemine ayrılır.
Saat dilimi semantiği ile zaman damgası
Spark, zaman damgalarını dahili olarak UTC değerleri olarak depolar ve belirli bir saat dilimi olmadan getirilen zaman damgası verileri yerel saat olarak UTC'ye mikrosaniye çözünürlükle dönüştürülür.
Zaman damgası verileri Spark'ta dışarı aktarıldığında veya görüntülendiğinde, zaman damgası değerlerini yerelleştirmek için oturum saat dilimi kullanılır. Oturum saat dilimi yapılandırmayla spark.sql.session.timeZone ayarlanır ve varsayılan olarak JVM sistemi yerel saat dilimine ayarlanır. pandas, nanosaniye çözünürlüğüne sahip ve sütun başına isteğe bağlı saat dilimi içerebilen bir datetime64 türü kullanırdatetime64[ns].
Zaman damgası verileri Spark'tan pandas'a aktarıldığında, nanosaniyeye dönüştürülür ve her sütun Spark oturum saat dilimine dönüştürülür, ardından bu saat dilimine yerelleştirilir, bu da saat dilimini kaldırır ve değerleri yerel saat olarak görüntüler.
toPandas() veya pandas_udf zaman damgası sütunlarıyla çağrıldığında bu durum oluşur.
Zaman damgası verileri pandas'tan Spark'a aktarıldığında UTC mikrosaniyesine dönüştürülür. Bu, pandas DataFrame ile çağrı createDataFrame yapılırken veya pandas UDF'den bir zaman damgası döndürülirken oluşur. Bu dönüştürmeler Spark'ın verileri beklenen biçimde olduğundan emin olmak için otomatik olarak yapılır, bu nedenle bu dönüştürmelerin hiçbirini kendiniz yapmanız gerekmez. Nanosaniye değerleri kesilir.
Standart bir UDF, zaman damgası verilerini Python datetime nesneleri olarak yükler ve bu da pandas zaman damgasından farklıdır. En iyi performansı elde etmek için pandas UDF'de zaman damgalarıyla çalışırken pandas zaman serisi işlevselliğini kullanmanızı öneririz. Ayrıntılar için bkz . Zaman Serisi / Tarih işlevselliği.
Örnek not defteri
Aşağıdaki not defteri, pandas UDF'leri ile başarabileceğiniz performans geliştirmelerini gösterir: