Pyspark 根据数据框 groupBy 制作多个文件

Posted 2023-04-17

【中文标题】Pyspark 根据数据框 groupBy 制作多个文件

【英文标题】：Pyspark make multiple files based on dataframe groupBy

【发布时间】：2020-07-29 04:43:58

【问题描述】：

我可以对大型数据集进行分组，并使用 Pandas 数据框制作多个 CSV、excel 文件。但是如何使用 Pyspark 数据框将 700K 记录分组到大约 230 个组中，并使 230 CSV 文件在国家/地区明智。 p>

使用熊猫

grouped = df.groupby("country_code")

# run this to generate separate Excel files
for country_code, group in grouped:
    group.to_excel(excel_writer=f"country_code.xlsx", sheet_name=country_code, index=False)

使用 Pyspark 数据框，当我尝试喜欢这个时-

for country_code, df_country in df.groupBy('country_code'):
    print(country_code,df_country.show(1))

它返回，

TypeError: 'GroupedData' 对象不可迭代

【参考方案1】：

如果您的要求是将所有国家/地区的数据保存在不同的文件中，您可以通过对数据进行分区来实现，但您将获得每个国家/地区的文件夹而不是文件，因为 spark 无法将数据直接保存到文件中。

每当调用数据帧编写器时，Spark 都会创建文件夹。

df.write.partitionBy('country_code').csv(path)

输出将是对应国家数据的多个文件夹

path/country_code=india/part-0000.csv
path/country_code=australia/part-0000.csv

如果您希望每个文件夹中有一个文件，您可以将数据重新分区为

df.repartition('country_code').write.partitionBy('country_code').csv(path)

【讨论】：

我的某些专栏包含数组数据结构，显示错误AnalysisException: CSV data source does not support array> 数据类型。; 在使用df.repartition('country_code').write.partitionBy('country_code').csv('grouped_data/') 之后，我想我需要将数组转换为字符串，然后再将partinionBy 转换为csv

您可以展开/展平数组或将数据保存为 json 或 parquet 文件

【参考方案2】：

在撰写本文时使用partitionBy，以便每个分区都基于您指定的列（在您的情况下为country_code）。

这是more。