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

Question

我有一个 DataFrame，其中有许多 str 类型的列，我想对所有这些列应用一个函数，而不重命名它们的名称或添加更多列，我尝试使用 for-in 循环执行 withColumn （见下面的例子），但通常当我运行代码时，它显示 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()

完整的错误很长，所以我决定只粘贴一些行。但是你可以在这里找到完整的跟踪 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.WhosebugError 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)。

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

Stack Overflow while processing several columns with a UDF

python

user-defined-functions

apache-spark

apache-spark-sql

pyspark