python，同时使用UDF处理多个列

Question

我有一个DataFrame，其中包含许多str类型的列，我想对所有这些列应用一个函数，而无需重命名它们的名称或添加更多列，我尝试使用执行withColumn的for-in循环（请参阅下面的示例），但通常当我运行代码时，它会显示一个Stack Overflow（它很少起作用）， 这个DataFrame一点也不大，它只有~15000条记录。

# df is a DataFrame
def lowerCase(string):
    return string.strip().lower()
lowerCaseUDF = udf(lowerCase, StringType())
for (columnName, kind) in df.dtypes:
    if(kind == "string"):
        df = df.withColumn(columnName, lowerCaseUDF(df[columnName]))
df.select("Tipo_unidad").distinct().show()

完整的错误很长，因此我决定只粘贴几行。但你可以在这里找到完整的跟踪 完整跟踪

Py4JJavaError： 调用 o516.showString 时出错。 ： org.apache.spark.SparkException：作业由于阶段故障而中止： 阶段 2.0 中的任务 1 失败了 4 次，最近一次失败：任务 1.3 丢失 在阶段 2.0 （TID 38， worker2.mcbo.mood.com.ve）： java.lang.StackOverflowError at java.io.ObjectInputStream$BlockDataInputStream.readByte（ObjectInputStream.java：2774）

我认为产生这个问题是因为此代码启动了许多作业（每列一个类型 string），您能否告诉我另一种选择或我做错了什么？

Answer 1

尝试这样的事情：

from pyspark.sql.functions import col, lower, trim
exprs = [
    lower(trim(col(c))).alias(c) if t == "string" else col(c) 
    for (c, t) in df.dtypes
]
df.select(*exprs)

与当前解决方案相比，此方法有两个主要优点：

• 它只需要作为单个投影（没有最有可能导致 SO 的增长谱系）而不是每个字符串列的投影。
• 它只直接操作内部表示，而不将数据传递给Python（BatchPythonProcessing）。