使用 UDF 处理多列时堆栈溢出 [英] Stack Overflow while processing several columns with a UDF
问题描述
我有一个 DataFrame
,其中包含许多 str
类型的列,我想对所有这些列应用一个函数,而不重命名它们的名称或添加更多列,我尝试使用 for-in
循环执行 withColumn
(参见下面的示例),但通常当我运行代码时,它会显示 Stack Overflow
(它很少工作),这个 DataFrame
一点也不大,它只有 ~15000 条记录.
I have a DataFrame
with many columns of str
type, and I want to apply a function to all those columns, without renaming their names or adding more columns, I tried using a for-in
loop executing withColumn
(see example bellow), but normally when I run the code, it shows a Stack Overflow
(it rarely works), this DataFrame
is not big at all, it has just ~15000 records.
# 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()
完整的错误很长,因此我决定只粘贴一些行.但您可以在此处找到完整的跟踪完整跟踪
The complete error is very long, therefore I decided to paste only some lines. But you can find the full trace here Complete Trace
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 在java.io.ObjectInputStream$BlockDataInputStream.readByte(ObjectInputStream.java:2774)
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
类型的每一列一个),你能告诉我另一种选择吗?或者我做错了什么?
I am thinking that this problem is produced because this code launches many jobs (one for each column of type string
), could you show me another alternative or what I am doing wrong?
推荐答案
试试这个:
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)
与您当前的解决方案相比,这种方法有两个主要优点:
This approach has two main advantages over you current solution:
- 它只需要作为单个投影(没有最有可能导致 SO 的不断增长的谱系)而不是每个字符串列的投影.
- 它仅直接操作内部表示而不将数据传递给 Python (
BatchPythonProcessing
).
这篇关于使用 UDF 处理多列时堆栈溢出的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!