PySpark如何在UDF中使用列名 按规则拼接值为正的列名

你代码的错误点 #

• 拼写错误：cols = inoutDF.columns里的inoutDF是笔误，应为inputDF
• UDF调用名称错误：你定义的UDF名为countUDF，调用时错写为countDF，名称不匹配导致报错
• 缺失拼接逻辑：仅完成了列值转列名/空串的转换步骤，没有处理空串过滤、最终列拼接的逻辑，直接拼接会出现多余逗号的问题

修复后的UDF实现方案 #

from pyspark.sql import functions as F
from pyspark.sql.types import StringType

# 原有UDF逻辑保留
def CountSelect(colname, x):
    return colname if x > 0 else ""

countUDF = F.udf(CountSelect, StringType())

inputDF = ... # 替换为你的原始DataFrame变量名
cols = inputDF.columns
cols.remove("ID")

# 生成中间转换列
intermediateDF = inputDF.select("ID", *(countUDF(c, F.col(c)).alias(c) for c in cols))

# 用concat_ws拼接自动忽略空字符串，避免多余逗号
resultDF = intermediateDF.select(
    "ID",
    F.concat_ws(",", *[F.col(c) for c in cols]).alias("NewColumn")
)

更简洁的无UDF实现方案（性能更优） #

无需自定义UDF，直接使用Spark内置函数实现，避免UDF带来的序列化性能开销：

from pyspark.sql import functions as F

inputDF = ... # 替换为你的原始DataFrame变量名
cols = inputDF.columns
cols.remove("ID")

resultDF = inputDF.select(
    "ID",
    F.concat_ws(
        ",",
        *[F.when(F.col(c) > 0, c) for c in cols]
    ).alias("NewColumn")
)

when条件不满足时默认返回null，concat_ws会自动跳过null值，最终拼接结果完全符合预期。

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