将 pyspark 数据帧与 groupBy 聚合后保存为 csv 文件

Question

我正在学习 pyspark，我对如何将分组数据帧保存为 csv 文件感到有点困惑（假设出于某些原因——例如 RAM 限制——我不想先将其转换为Pandas 数据框）。

一个可重现的例子：

import seaborn as sns
import findspark
findspark.init()
import pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.master('local') \
.appName('Data cleaning') \
.getOrCreate()
from pyspark.context import SparkContext
sc = SparkContext.getOrCreate()
from pyspark.sql.functions import *

mpg= sns.load_dataset('mpg')
mpg_sp = spark.createDataFrame(mpg)
mpg_grp = mpg_sp.groupBy('model_year', 'origin').avg('displacement', 'weight')

# The command below fails in the sense that it creates a folder with multiple  files in it rather than a single csv file as I would expect

mpg_grp.write.csv('mpg_grp.csv')

# By applying the collect method I get a list which can not be saved as a csv file

mpg_grp1 = mpg_grp.collect()
type(mpg_grp1)
list

Answer 1

Spark 是一个分布式框架。因此，多个文件中的输出是一种正常行为......每个工作人员都会编写它的一部分，从而产生几个小文件。

您可以使用此命令来欺骗系统：

mpg_grp.coalesce(1).write.csv('mpg_grp.csv')

这只会写入 1 个文件（但仍在名称为 'mpg_grp.csv' 的文件夹中）。
注意：可能会很慢。

Answer 2

上面的答案是正确的，但使用效果不是很好。
当然，您可以使用 repartition(1) 或 coalesce(1)，但这会导致将所有数据传输给单个 worker，并会大大降低您的代码速度。
为了避免这种情况，我建议您在数据集中的一列上对数据进行分区。然后写个简单的代码让每个分区一个文件：

cols = ["$name"]
mpg_grp.repartition(cols).write.partitionBy(cols).csv("$location")

因此，数据将按您的其中一列在工作人员之间进行分区，并且您将在每个分区中得到一个文件（例如按日期）。

将 pyspark 数据帧与 groupBy 聚合后保存为 csv 文件

Saving pyspark dataframe after being aggregated with groupBy as csv file

python

pandas

pyspark

pyspark-sql