PySpark UDF执行报TypeError需使用lit/array等函数问题求解

报错核心是你直接将Python原生的优先级列表作为参数传递给了PySpark UDF：

• PySpark执行UDF时，所有传入UDF的非列类型常量都必须用lit()/array()等Spark内置函数包装为列字面量，才能正常序列化后传递给Python worker执行，原生Python list类型无法被Spark直接识别
• 纯Python环境没有Spark的参数序列化校验逻辑，直接传递原生列表不会触发问题，因此会出现本地测试正常、PySpark执行报错的差异

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf

priority_list = [79, -1, -1] # 原生Python列表，未做Spark类型转换

@udf(returnType="integer")
def extract_highest_priority(event_str, priority):
    event_list = [int(i) for i in event_str.split(",")]
    for p in priority:
        if p in event_list:
            return p
    return None

# 直接传入原生列表作为UDF参数，触发报错
df = df.withColumn("result", extract_highest_priority("event_list", priority_list))

根据优先级数组是否需要动态调整，对应两种处理方案：

方案1：优先级数组为固定常量 #

直接将优先级数组定义在UDF内部，无需作为参数传入：

@udf(returnType="integer")
def extract_highest_priority(event_str):
    # 固定优先级直接写在UDF内部，避免参数传递
    priority = [79, -1, -1]
    if not event_str:
        return None
    event_list = [int(i) for i in event_str.split(",")]
    for p in priority:
        if p in event_list:
            return p
    return None

# 调用仅需传入event_list列
df = df.withColumn("result", extract_highest_priority("event_list"))

方案2：优先级数组需要动态传入 #

用array()+lit()将原生Python列表包装为Spark Array类型列字面量后再传入UDF：

from pyspark.sql.functions import array, lit

priority_list = [79, -1, -1]
# 原生列表转Spark列字面量
spark_priority = array(*[lit(x) for x in priority_list])

@udf(returnType="integer")
def extract_highest_priority(event_str, priority):
    if not event_str:
        return None
    event_list = [int(i) for i in event_str.split(",")]
    for p in priority:
        if p in event_list:
            return p
    return None

# 传入包装后的Spark类型优先级参数
df = df.withColumn("result", extract_highest_priority("event_list", spark_priority))

• 建议在UDF中增加空值、非法字符串格式的异常兼容逻辑，避免运行时出现类型转换、空指针错误
• 全量执行前可先用小批量样本验证UDF输出是否符合预期

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