How do you control the size of the output file?

by

It’s impossible for Spark to control the size of Parquet files, because the DataFrame in memory needs to be encoded and compressed before writing to disks. Before this process finishes, there is no way to estimate the actual file size on disk.

So my solution is:

  • Write the DataFrame to HDFS, df.write.parquet(path)

  • Get the directory size and calculate the number of files

    val fs = FileSystem.get(sc.hadoopConfiguration)
val dirSize = fs.getContentSummary(path).getLength
val fileNum = dirSize/(512 * 1024 * 1024)  // let's say 512 MB per file

  • Read the directory and re-write to HDFS

    val df = sqlContext.read.parquet(path)
df.coalesce(fileNum).write.parquet(another_path)

    Do NOT reuse the original df, otherwise it will trigger your job two times.

  • Delete the old directory and rename the new directory back

    fs.delete(new Path(path), true)
fs.rename(new Path(newPath), new Path(path))

This solution has a drawback that it needs to write the data two times, which doubles disk IO, but for now this is the only solution.

More Related Contents:

  1. Spark Dataframe validating column names for parquet writes
  2. Why are Spark Parquet files for an aggregate larger than the original?
  3. Spark lists all leaf node even in partitioned data
  4. How to optimize partitioning when migrating data from JDBC source?
  5. Spark: subtract two DataFrames
  6. TypeError: Column is not iterable – How to iterate over ArrayType()?
  7. How to overwrite the output directory in spark
  8. Is groupByKey ever preferred over reduceByKey
  9. Reading DataFrame from partitioned parquet file
  10. Pyspark : forward fill with last observation for a DataFrame
  11. spark.sql.crossJoin.enabled for Spark 2.x
  12. Spark DataFrame Schema Nullable Fields
  13. PySpark: how to resample frequencies
  14. Explode array data into rows in spark [duplicate]
  15. Why does Spark think this is a cross / Cartesian join
  16. Why does sortBy transformation trigger a Spark job?
  17. Spark SQL broadcast hash join
  18. What is the difference between cache and persist?
  19. What is the difference between Apache Spark SQLContext vs HiveContext?
  20. Saving dataframe to local file system results in empty results
  21. Dealing with a large gzipped file in Spark
  22. When are accumulators truly reliable?
  23. What should be the optimal value for spark.sql.shuffle.partitions or how do we increase partitions when using Spark SQL?
  24. pyspark: Efficiently have partitionBy write to same number of total partitions as original table
  25. How to get day of week in SparkSQL?
  26. AWS EMR – ModuleNotFoundError: No module named ‘pyarrow’
  27. PySpark: How to fillna values in dataframe for specific columns?
  28. How to calculate Median in spark sqlContext for column of data type double
  29. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’
  30. Understanding Spark’s caching

Leave a Comment