在 pyspark 中对大量列进行累积求和的优化方法

Posted 2023-04-15

【中文标题】在 pyspark 中对大量列进行累积求和的优化方法

【英文标题】：Optimised way of doing cumulative sum on large number of columns in pyspark

【发布时间】：2018-12-18 18:56:31

【问题描述】：

我有一个包含 752 个（id、date 和 750 个特征列）列和大约 150 万行的 DataFrame，我需要对所有 750 个按 id 分区和按日期排序的特征列应用累积总和。

以下是我目前正在遵循的方法：

# putting all 750 feature columns in a list
required_columns = ['ts_1','ts_2'....,'ts_750']

# defining window
sumwindow = Window.partitionBy('id').orderBy('date')

# Applying window to calculate cumulative of each individual feature column

for current_col in required_columns:
    new_col_name = "sum_0".format(current_col)
    df=df.withColumn(new_col_name,sum(col(current_col)).over(sumwindow))

# Saving the result into parquet file    
df.write.format('parquet').save(output_path)

我在运行当前方法时遇到错误

py4j.protocol.Py4JJavaError: An error occurred while calling o2428.save.
: java.lang.***Error

请让我知道相同的替代解决方案。对于大量数据，累积总和似乎有点棘手。请建议任何替代方法或任何我可以调整以使其工作的火花配置。

【问题讨论】：

您当前的窗口不会做累计和 - 它只是所有行的总和。请查看this post，了解如何进行累积和的示例。

嘿 pault 实际上我想按 id 对每一列的累积总和进行分组（它不是唯一的）。窗口工作正常我已经用较小的数据集进行了测试。它只是给我大数据集的***错误

【参考方案1】：

我想你有过大的血统的问题。在您多次重新分配数据框后，请查看您的解释计划。

对此的标准解决方案是每隔一段时间检查一次您的数据帧以截断解释计划。这有点像缓存，但用于计划而不是数据，并且经常需要用于修改数据帧的迭代算法。

Here 是对缓存和检查点的一个很好的 pyspark 解释

我建议 df.checkpoint() 每修改 5-10 次就开始

告诉我们进展如何

【讨论】：

您可以尝试其中一些技巧来缩短时间。不确定它们是否适合您的用例。 ***.com/questions/33882894/…