使用UDF处理多个列时堆栈溢出

Question

我有一个DataFrame，它有很多str类型的列，我想对所有这些列应用一个函数，不需要重命名它们的名称或添加更多的列，我尝试使用一个执行withColumn的for-in循环（见下面的示例），但通常当我运行代码时，它显示一个Stack Overflow（它很少起作用），这个{}一点也不大，它只有大约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()

完全错误很长，所以我决定只粘贴一些行。但是你可以在这里找到完整的轨迹Complete Trace

Py4JJavaError: An error occurred while calling o516.showString. : org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 2.0 failed 4 times, most recent failure: Lost task 1.3 in stage 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)

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

• 它只需要作为单个投影（没有最有可能导致这种情况的增长血统）而不是每个字符串列的投影。在
• 它只直接操作内部表示而不向Python（BatchPythonProcessing）传递数据。在