创建含pd.Series与行内参数的Pandas UDF遇签名不支持错误

你的问题核心在于Pandas UDF的参数类型定义错误，以及返回值类型不匹配：

1. 参数类型问题：Spark的Pandas UDF中，每个输入参数对应整个DataFrame列的pd.Series，而非单行的标量值。你把addition参数标注为int，但实际Spark会传入该列所有行组成的pd.Series，这直接触发了签名不支持的异常。
2. 返回值类型问题：你的UDF返回的是数组，但标注的返回类型是IntegerType()，与实际返回值类型不匹配。

修正后的代码示例 #

from pyspark.sql.types import ArrayType, IntegerType
import pyspark.sql.functions as F
import pandas as pd

@F.pandas_udf(ArrayType(IntegerType()))
def addition(arr_series: pd.Series, addition_series: pd.Series) -> pd.Series:
    # 逐行配对数组和对应的加数，执行计算
    return pd.Series([
        [num + add for num in arr] 
        for arr, add in zip(arr_series, addition_series)
    ])

# 测试数据
df = spark.createDataFrame([([1,2,3],10),([4,5,6],20)],["array","addition"])
df.show()

# 调用UDF生成新列
df.withColumn("added", addition(F.col("array"), F.col("addition"))).show()

复杂逻辑的适配方式 #

如果真实业务逻辑复杂，可以在UDF内部对每行的数组单独处理（比如用pd.Series做复杂运算），示例如下：

@F.pandas_udf(ArrayType(IntegerType()))
def complex_row_calc(arr_series: pd.Series, param_series: pd.Series) -> pd.Series:
    def process_single_row(arr, param):
        # 这里替换为你的复杂逻辑
        s = pd.Series(arr)
        # 示例：过滤大于1的元素，乘以参数后加5
        processed = s[s > 1] * param + 5
        return processed.tolist()
    
    # 逐行调用处理函数
    return pd.Series([process_single_row(arr, p) for arr, p in zip(arr_series, param_series)])

关键注意事项 #

• 所有输入参数必须标注为pd.Series，对应Spark DataFrame的整列数据
• UDF的返回类型必须与实际返回值的结构匹配（比如数组对应ArrayType）
• 逐行处理时，用zip配对多列的对应行元素，确保同一行的参数一起计算

内容的提问来源于stack exchange，提问作者Wim Schmitz