register (UDFRegistration)

bir Python işlevini (lambda işlevleri dahil) veya kullanıcı tanımlı bir işlevi SQL işlevi olarak kaydeder.

Sözdizimi

register(name, f, returnType=None)

Parametreler

Parametre	Türü	Açıklama
name	str	SQL deyimlerindeki kullanıcı tanımlı işlevin adı.
f	function, udf, veya pandas_udf	Python işlevi veya kullanıcı tanımlı bir işlev. Kullanıcı tanımlı işlev bir kerede satır veya vektörleştirilebilir.
returnType	DataType veya str, isteğe bağlı	Kayıtlı kullanıcı tanımlı işlevin dönüş türü. Nesne DataType veya DDL biçimli bir tür dizesi olabilir. f zaten kullanıcı tanımlı bir işlev olduğunda değil, yalnızca f düz Python bir işlev olduğunda geçerlidir.

İadeler

function

Notlar

Belirsiz bir Python işlevini kaydetmek için, önce Python işlevi için tanımsız bir kullanıcı tanımlı işlev oluşturun ve sonra bunu sql işlevi olarak kaydedin.

Örnekler

# Register a lambda as a SQL function (return type defaults to string).
strlen = spark.udf.register("stringLengthString", lambda x: len(x))
spark.sql("SELECT stringLengthString('test')").collect()
# [Row(stringLengthString(test)='4')]

spark.sql("SELECT 'foo' AS text").select(strlen("text")).collect()
# [Row(stringLengthString(text)='3')]

# Register with an explicit return type.
from pyspark.sql.types import IntegerType
spark.udf.register("stringLengthInt", lambda x: len(x), IntegerType())
spark.sql("SELECT stringLengthInt('test')").collect()
# [Row(stringLengthInt(test)=4)]

# Register an existing UDF.
from pyspark.sql.functions import udf
slen = udf(lambda s: len(s), IntegerType())
spark.udf.register("slen", slen)
spark.sql("SELECT slen('test')").collect()
# [Row(slen(test)=4)]

# Register a nondeterministic UDF.
import random
random_udf = udf(lambda: random.randint(0, 100), IntegerType()).asNondeterministic()
spark.udf.register("random_udf", random_udf)

# Register a pandas UDF.
import pandas as pd
from pyspark.sql.functions import pandas_udf

@pandas_udf("integer")
def add_one(s: pd.Series) -> pd.Series:
    return s + 1

spark.udf.register("add_one", add_one)
spark.sql("SELECT add_one(id) FROM range(3)").collect()
# [Row(add_one(id)=1), Row(add_one(id)=2), Row(add_one(id)=3)]

# Register a grouped aggregate pandas UDF.
@pandas_udf("integer")
def sum_udf(v: pd.Series) -> int:
    return v.sum()

spark.udf.register("sum_udf", sum_udf)
spark.sql(
    "SELECT sum_udf(v1) FROM VALUES (3, 0), (2, 0), (1, 1) tbl(v1, v2) GROUP BY v2"
).sort("sum_udf(v1)").collect()
# [Row(sum_udf(v1)=1), Row(sum_udf(v1)=5)]