Pyspark：使用 Window 函数按日期将数据框保存到单个 csv？

Question

我有一个这样的数据框：

df = pd.DataFrame({"Date": ["2020-05-10", "2020-05-10", "2020-05-10", "2020-05-11", "2020-05-11", "2020-05-11", "2020-05-11", "2020-05-11", "2020-05-11"],
                   "Slot_Length": [30, 30, 30, 30, 30, 30, 30, 30, 30],
                   "Total_Space": [60, 60, 60, 120, 120, 120, 120, 120, 120],
                   "Amount_Over": [-30, -30, -30, -60, -60, -60, -60, -60, -60],
                   "Rank": [1, 1, 2, 1, 1, 1, 1, 2, 2]})

df = spark.createDataFrame(df)

+----------+-----------+-----------+-----------+----+
|      Date|Slot_Length|Total_Space|Amount_Over|Rank|
+----------+-----------+-----------+-----------+----+
|2020-05-10|         30|         60|        -30|   1|
|2020-05-10|         30|         60|        -30|   1|
|2020-05-10|         30|         60|        -30|   2|
|2020-05-11|         30|        120|        -60|   1|
|2020-05-11|         30|        120|        -60|   1|
|2020-05-11|         30|        120|        -60|   1|
|2020-05-11|         30|        120|        -60|   1|
|2020-05-11|         30|        120|        -60|   2|
|2020-05-11|         30|        120|        -60|   2|
+----------+-----------+-----------+-----------+----+

我知道我可以像这样将数据帧保存到单个 csv 文件中：

df.coalesce(1).write.format("com.databricks.spark.csv"
                                       ).mode('overwrite'
                                             ).option("header", "true"
                                               ).save("s3://mycsv_date.csv")

我想按日期拆分数据框并将每个过滤后的数据框保存到 csv。

mycsv_2020_05_10.csv
mycsv_2020_05_11.csv

最好的方法是什么？

Answer 1

使用

df.repartition('Date').write.partitionBy('Date').format("com.databricks.spark.csv"
                                       ).mode('overwrite'
                                             ).option("header", "true"
                                               ).save("s3://bucket/path")

现在您将在每个分区中拥有每个日期的文件夹和单个文件

Pyspark：使用 Window 函数按日期将数据框保存到单个 csv？

Pyspark: Save dataframe to single csv by date using Window function?

apache-spark-sql

pyspark

pyspark-dataframes