Why are Spark Parquet files for an aggregate larger than the original?

by

In general columnar storage formats like Parquet are highly sensitive when it comes to data distribution (data organization) and cardinality of individual columns. The more organized is data and the lower is cardinality the more efficient is the storage.

Aggregation, as the one you apply, has to shuffle the data. When you check the execution plan you’ll see it is using hash partitioner. It means that after aggregation distribution can be less efficient than the one for the original data. At the same time sum can reduce number of rows but increase cardinality for rCount column.

You can try different tools to correct for that but not all are available in Spark 1.5.2:

  • Sort complete dataset by columns having low cardinality (quite expensive due to full shuffle) or sortWithinPartitions.
  • Use partitionBy method of DataFrameWriter to partition data using low cardinality columns.
  • Use bucketBy and sortBy methods of DataFrameWriter (Spark 2.0.0+) to improve data distribution using bucketing and local sorting.

More Related Contents:

  1. Spark Dataframe validating column names for parquet writes
  2. How do you control the size of the output file?
  3. Spark lists all leaf node even in partitioned data
  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. Integrating Spark Structured Streaming with the Confluent Schema Registry
  8. While writing to hdfs path getting error java.io.IOException: Failed to rename
  9. Multiple Aggregate operations on the same column of a spark dataframe
  10. Partitioning in spark while reading from RDBMS via JDBC
  11. How to read multiple text files into a single RDD?
  12. How to handle categorical features with spark-ml?
  13. How to access element of a VectorUDT column in a Spark DataFrame?
  14. Python Spark Cumulative Sum by Group Using DataFrame
  15. Link Spark with iPython Notebook
  16. Default Partitioning Scheme in Spark
  17. Any performance issues forcing eager evaluation using count in spark?
  18. How does createOrReplaceTempView work in Spark?
  19. Debugging “Managed memory leak detected” in Spark 1.6.0
  20. Spark: how to get the number of written rows?
  21. Save ML model for future usage
  22. Determining optimal number of Spark partitions based on workers, cores and DataFrame size
  23. Keep only duplicates from a DataFrame regarding some field
  24. What’s the difference between Spark ML and MLLIB packages
  25. Overwrite only some partitions in a partitioned spark Dataset
  26. PySpark: How to fillna values in dataframe for specific columns?
  27. How to optimize shuffle spill in Apache Spark application
  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