DataFrame partitionBy to a single Parquet file (per partition)

by

I had the exact same problem and I found a way to do this using DataFrame.repartition(). The problem with using coalesce(1) is that your parallelism drops to 1, and it can be slow at best and error out at worst. Increasing that number doesn’t help either — if you do coalesce(10) you get more parallelism, but end up with 10 files per partition.

To get one file per partition without using coalesce(), use repartition() with the same columns you want the output to be partitioned by. So in your case, do this:

import spark.implicits._
df
  .repartition($"entity", $"year", $"month", $"day", $"status")
  .write
  .partitionBy("entity", "year", "month", "day", "status")
  .mode(SaveMode.Append)
  .parquet(s"$location")

Once I do that I get one parquet file per output partition, instead of multiple files.

I tested this in Python, but I assume in Scala it should be the same.

More Related Contents:

  1. Finding duplicates from large data set using Apache Spark
  2. Using a column value as a parameter to a spark DataFrame function
  3. Find maximum row per group in Spark DataFrame
  4. Split Spark Dataframe string column into multiple columns
  5. While writing to hdfs path getting error java.io.IOException: Failed to rename
  6. Multiple Aggregate operations on the same column of a spark dataframe
  7. Partitioning in spark while reading from RDBMS via JDBC
  8. How to control partition size in Spark SQL
  9. How to import multiple csv files in a single load?
  10. How to access element of a VectorUDT column in a Spark DataFrame?
  11. Spark DataFrame: count distinct values of every column
  12. How to split a list to multiple columns in Pyspark?
  13. How to save/insert each DStream into a permanent table
  14. Spark load data and add filename as dataframe column
  15. What is the maximum size for a broadcast object in Spark?
  16. PySpark: how to resample frequencies
  17. How to find mean of grouped Vector columns in Spark SQL?
  18. How does createOrReplaceTempView work in Spark?
  19. Fill in null with previously known good value with pyspark
  20. PySpark – get row number for each row in a group
  21. How to find count of Null and Nan values for each column in a PySpark dataframe efficiently?
  22. pyspark: count distinct over a window
  23. Using windowing functions in Spark
  24. PySpark: How to fillna values in dataframe for specific columns?
  25. Why spark application fail with “executor.CoarseGrainedExecutorBackend: Driver Disassociated”?
  26. What does “Correlated scalar subqueries must be Aggregated” mean?
  27. How to improve broadcast Join speed with between condition in Spark
  28. How to calculate Median in spark sqlContext for column of data type double
  29. Spark SQL referencing attributes of UDT
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment