Spark parquet partitioning : Large number of files

by

First I would really avoid using coalesce, as this is often pushed up further in the chain of transformation and may destroy the parallelism of your job (I asked about this issue here : Coalesce reduces parallelism of entire stage (spark))

Writing 1 file per parquet-partition is realtively easy (see Spark dataframe write method writing many small files):

data.repartition($"key").write.partitionBy("key").parquet("/location")

If you want to set an arbitrary number of files (or files which have all the same size), you need to further repartition your data using another attribute which could be used (I cannot tell you what this might be in your case):

data.repartition($"key",$"another_key").write.partitionBy("key").parquet("/location")

another_key could be another attribute of your dataset, or a derived attribute using some modulo or rounding-operations on existing attributes. You could even use window-functions with row_number over key and then round this by something like

data.repartition($"key",floor($"row_number"/N)*N).write.partitionBy("key").parquet("/location")

This would put you N records into 1 parquet file

using orderBy

You can also control the number of files without repartitioning by ordering your dataframe accordingly:

data.orderBy($"key").write.partitionBy("key").parquet("/location")

This will lead to a total of (at least, but not much more than) spark.sql.shuffle.partitions files across all partitions (by default 200). It’s even beneficial to add a second ordering column after $key, as parquet will remember the ordering of the dataframe and will write the statistics accordingly. For example, you can order by an ID:

data.orderBy($"key",$"id").write.partitionBy("key").parquet("/location")

This will not change the number of files, but it will improve the performance when you query your parquet file for a given key and id. See e.g. https://www.slideshare.net/RyanBlue3/parquet-performance-tuning-the-missing-guide and https://db-blog.web.cern.ch/blog/luca-canali/2017-06-diving-spark-and-parquet-workloads-example

Spark 2.2+

From Spark 2.2 on, you can also play with the new option maxRecordsPerFile to limit the number of records per file if you have too large files. You will still get at least N files if you have N partitions, but you can split the file written by 1 partition (task) into smaller chunks:

df.write
.option("maxRecordsPerFile", 10000)
...

See e.g. http://www.gatorsmile.io/anticipated-feature-in-spark-2-2-max-records-written-per-file/ and spark write to disk with N files less than N partitions

More Related Contents:

  1. Determining optimal number of Spark partitions based on workers, cores and DataFrame size
  2. Apache spark dealing with case statements
  3. Unable to fetch the value of Println in apache spark
  4. Finding duplicates from large data set using Apache Spark
  5. Spark – repartition() vs coalesce()
  6. How do I split an RDD into two or more RDDs?
  7. What does “Stage Skipped” mean in Apache Spark web UI?
  8. Overwrite specific partitions in spark dataframe write method
  9. How to import multiple csv files in a single load?
  10. Spark: subtract two DataFrames
  11. How to improve performance for slow Spark jobs using DataFrame and JDBC connection?
  12. TypeError: Column is not iterable – How to iterate over ArrayType()?
  13. Which operations preserve RDD order?
  14. Is groupByKey ever preferred over reduceByKey
  15. Python Spark Cumulative Sum by Group Using DataFrame
  16. How to save/insert each DStream into a permanent table
  17. Default Partitioning Scheme in Spark
  18. How DAG works under the covers in RDD?
  19. Apache Spark: What is the equivalent implementation of RDD.groupByKey() using RDD.aggregateByKey()?
  20. Why does sortBy transformation trigger a Spark job?
  21. PySpark DataFrames – way to enumerate without converting to Pandas?
  22. Pyspark: Pass multiple columns in UDF
  23. What is the difference between cache and persist?
  24. How spark read a large file (petabyte) when file can not be fit in spark’s main memory
  25. Does a join of co-partitioned RDDs cause a shuffle in Apache Spark?
  26. Spark ALS predictAll returns empty
  27. Why is the fold action necessary in Spark?
  28. Stackoverflow due to long RDD Lineage
  29. Spark: Reading files using different delimiter than new line
  30. Convert null values to empty array in Spark DataFrame

Leave a Comment