PySpark UDF执行触发TypeError：min函数参数异常求助

定义了如下Python函数用于计算EAD：

def ead(lista):
    ind_mmff, isdebala, isfubala, k1, k2, ead = lista
    try:
        isdebala = float(isdebala)
        isfubala = float(isfubala)
        k1 = float(k1)
        k2 = float(k2)
        ead = float(ead)
    except ValueError:
        return 'Error: invalid input'        
    min_deb = min(0, isdebala)
    min_fub = min(0, isfubala)
    
    if ind_mmff == '0':
        ead_dai = abs(min_deb * k1 / 100 + min_fub * k2 / 100)
    else:
        ead_dai = ead
    return ead_dai

将其注册为Spark UDF并调用：

ead_udf = udf(lambda z: ead(z), FloatType())
df = df.withColumn('ead_calc', ead_udf(array(df.ind_mmff, df.isdebala, df.isfubala, df.k1, df.k2, df.ead_final_motor)))

执行df.select('ead_calc').show()时触发错误，错误定位在min_deb = min(0, isdebala)行，提示：

TypeError: _() takes 1 positional argument but 2 were given

核心问题是函数内部的变量名与函数名重名：函数名为ead，但在函数内部又将lista的第六个元素赋值给了同名变量ead，这会导致Python命名空间中局部变量ead覆盖了函数自身的引用，进而在Spark UDF的执行环境中引发命名冲突，间接破坏内置min函数的正常调用逻辑，出现参数数量不匹配的错误。

此外，函数可能返回字符串类型的错误信息，但UDF声明的返回类型是FloatType()，类型不匹配也会埋下潜在隐患。

1. 修改变量名，避免与函数名冲突 #

将函数内部的ead变量替换为其他名称（如ead_val），消除命名冲突：

def ead(lista):
    ind_mmff, isdebala, isfubala, k1, k2, ead_val = lista
    try:
        isdebala = float(isdebala)
        isfubala = float(isfubala)
        k1 = float(k1)
        k2 = float(k2)
        ead_val = float(ead_val)
    except ValueError:
        return None  # 改为返回None适配FloatType，或保留字符串并修改UDF返回类型
    min_deb = min(0, isdebala)
    min_fub = min(0, isfubala)
    
    if ind_mmff == '0':
        ead_dai = abs(min_deb * k1 / 100 + min_fub * k2 / 100)
    else:
        ead_dai = ead_val
    return ead_dai

2. 修正UDF返回类型 #

如果选择返回None表示错误，保持FloatType()即可；如果需要保留错误字符串提示，将UDF返回类型改为StringType()：

# 情况1：返回数值或None，用FloatType
ead_udf = udf(ead, FloatType())

# 情况2：返回字符串错误信息，用StringType
# ead_udf = udf(ead, StringType())

3. 简化UDF注册（可选） #

无需用lambda包裹，直接将函数传递给udf：

ead_udf = udf(ead, FloatType())

4. 调用验证 #

重新生成ead_calc列并验证：

df = df.withColumn('ead_calc', ead_udf(array(df.ind_mmff, df.isdebala, df.isfubala, df.k1, df.k2, df.ead_final_motor)))
df.select('ead_calc').show()

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