Why does Spark writes dataframe to different file sizes and different time?

user_1251549 0
Viewed 165

Consider general algorithm:

val first : DataFrame = ...            // about 100 Mb
val second : DataFrame = ...           // about 5 GMb 
val third : DataFrame = ...            // about 7 GMb
val fourth : DataFrame = ...           // about 13 GMb

//all dataframe are filtered, renamed all columns. A new colun is added into `third` and `fourth`  

val = firstAndSecond = first.join(first("first_id") === second("second_id"))
val = thirdsAndAll = firstAndSecond.join(firstAndSecond("some_id") === third("third_id"))
val = fourthAndAll = thirdsAndAll.join(thirdsAndAll("other_id") === fourth("fourth_id"))

fourthAndAll.write.mode(saveMode = SaveMode.Overwrite).parquet("file://C:path")

Notes*

  1. All data frames are read to and written from SSD drive.
  2. Read and write operation performed to/in parquet files
  3. Program was run on to Threadripper with 8 cores (16 virtual), 80Gb Ram (spark consumed about 25Gb), also 99% of time (except the situation when last file is writeen) all 16 cores are loaded 100%.

The problem

I have very different sizes in output parquet files from 100 kb to 500Nb. Also bigger files have very long time to write. E.g. Every file is writen by some threads which wrote 500Mb, 450Mb (etc.) files perform their work too long. (for 500Mb it was about 8 hours)

Any thoughts how to setup spark to write parquet file with more or less equal size CPU load?

apache-spark scala
1 Answers

(I work mostly with RDDs, have not much experience with dataframe api but I guess it works the same way)

I guess .parquet creates a file per partition of the rdd. If file sizes are drastically different then the data is not evenly distributed across partitions. It also could decrease the performance of join operation. repartition command should help.

user_53850410
Related