在 Pyspark 中应用 partitionBy 后如何拥有单个 csv 文件
How to have a single csv file after applying partitionBy in Pysark
我必须首先按“客户组”进行分区，但我还想确保每个“customer_group”都有一个 csv 文件。这是因为推理需要的是时间序列数据，不能分布在多个文件中。
我试过：datasink2 = spark_df1.write.format("csv").partitionBy('customer_group').option("compression","gzip").save(destination_path+'/traintestcsvzippartitionocalesce')
但它会在 customer_group/ 路径中创建多个较小的文件，格式为 csv.gz0000_part_00.gz , csv.gz0000_part_01.gz ....
我尝试使用：datasink2 = spark_df1.write.format("csv").partitionBy('customer_group').coalesce(1).option("compression","gzip").save(destination_path+'/traintestcsvzippartitionocalesce')
但它会引发以下错误：
AttributeError: 'DataFrameWriter' 对象没有属性 'coalesce'
有解决办法吗？
我不能在没有分区的情况下直接使用 repartition(1) 或 coalesce(1)，因为它只创建 1 个文件并且一次只有一个工作节点工作（串行）并且计算量非常大
repartition 函数也接受列名作为参数，而不仅仅是分区数。
按写入分区列重新分区将使spark每个文件夹保存一个文件。
请注意，如果您的一个分区出现偏差并且一个客户组拥有大部分数据，您可能会遇到性能问题。
spark_df1 \
.repartition("customer_group") \
.write \
.partitionBy("customer_group") \
...
我必须首先按“客户组”进行分区，但我还想确保每个“customer_group”都有一个 csv 文件。这是因为推理需要的是时间序列数据，不能分布在多个文件中。
我试过：datasink2 = spark_df1.write.format("csv").partitionBy('customer_group').option("compression","gzip").save(destination_path+'/traintestcsvzippartitionocalesce')
但它会在 customer_group/ 路径中创建多个较小的文件，格式为 csv.gz0000_part_00.gz , csv.gz0000_part_01.gz ....
我尝试使用：datasink2 = spark_df1.write.format("csv").partitionBy('customer_group').coalesce(1).option("compression","gzip").save(destination_path+'/traintestcsvzippartitionocalesce')
但它会引发以下错误：
AttributeError: 'DataFrameWriter' 对象没有属性 'coalesce'
有解决办法吗？
我不能在没有分区的情况下直接使用 repartition(1) 或 coalesce(1)，因为它只创建 1 个文件并且一次只有一个工作节点工作（串行）并且计算量非常大
repartition 函数也接受列名作为参数，而不仅仅是分区数。 按写入分区列重新分区将使spark每个文件夹保存一个文件。
请注意，如果您的一个分区出现偏差并且一个客户组拥有大部分数据，您可能会遇到性能问题。
spark_df1 \
.repartition("customer_group") \
.write \
.partitionBy("customer_group") \
...