PySpark UDF异常：参数值被意外追加至DataFrame列中

你的问题出在partial绑定参数的顺序与函数定义不匹配，导致UDF执行时参数完全颠倒：

• process_lower_function定义的参数顺序是(text, lower_rule)，其中text是DataFrame列的输入值，lower_rule是规则字符串
• 但你用partial(process_lower_function, process_rule)时，是把process_rule绑定到了函数的第一个参数（也就是text的位置），而UDF传入的"data"列值被传到了第二个参数lower_rule的位置
• 这就导致函数里实际是把规则字符串当作待处理文本，把列值当作规则去匹配，最终出现不符合预期的结果

方案1：用关键字参数绑定partial（推荐，避免顺序问题） #

直接通过关键字参数指定要绑定的lower_rule，彻底规避参数顺序错误：

#!/usr/bin/env python
import sys
import logging
from pyspark.sql.types import StringType
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from functools import partial

def main():
    spark = SparkSession.builder.master("local[1]").appName('intelligent').getOrCreate()
    data = [(0, "cards"), (1, "upper")]
    deptColumns = ["index", "data"]
    df = spark.createDataFrame(data=data, schema = deptColumns)
    print(df.show())
    process_rule = "pyspark,testing,upper"
    # 用关键字参数绑定lower_rule，不受参数顺序影响
    lower_function = partial(process_lower_function, lower_rule=process_rule)
    # 明确指定UDF返回类型，避免Spark自动推断出错
    udf_parser = udf(lower_function, StringType())
    df = df.withColumn("ndata", udf_parser("data"))
    print(df.show())


def process_lower_function(text, lower_rule):
    for rule in lower_rule.split(","):
        if rule in text:
            text = text.replace(rule, rule.upper())
            logging.info(f"LOWER :: {text} -- {rule}")
    return text


if __name__ == "__main__":
    sys.exit(main())

方案2：调整函数参数顺序 #

把lower_rule放在函数的第一个参数位置，让partial的绑定逻辑匹配预期：

#!/usr/bin/env python
import sys
import logging
from pyspark.sql.types import StringType
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from functools import partial

def main():
    spark = SparkSession.builder.master("local[1]").appName('intelligent').getOrCreate()
    data = [(0, "cards"), (1, "upper")]
    deptColumns = ["index", "data"]
    df = spark.createDataFrame(data=data, schema = deptColumns)
    print(df.show())
    process_rule = "pyspark,testing,upper"
    lower_function = partial(process_lower_function, process_rule)
    udf_parser = udf(lower_function, StringType())
    df = df.withColumn("ndata", udf_parser("data"))
    print(df.show())


# 调整参数顺序，lower_rule放在第一个位置
def process_lower_function(lower_rule, text):
    for rule in lower_rule.split(","):
        if rule in text:
            text = text.replace(rule, rule.upper())
            logging.info(f"LOWER :: {text} -- {rule}")
    return text


if __name__ == "__main__":
    sys.exit(main())

额外优化点 #

• 给UDF指定明确的返回类型（StringType()），避免Spark因类型推断出现意外问题
• 代码中未使用的process_function字典可以直接删除，精简代码

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