5

I am trying to use Spark's bucketBy feature on a pretty large dataset.

dataframe.write()
    .format("parquet")
    .bucketBy(500, bucketColumn1, bucketColumn2)
    .mode(SaveMode.Overwrite)
    .option("path", "s3://my-bucket")
    .saveAsTable("my_table");

The problem is that my Spark cluster has about 500 partitions/tasks/executors (not sure the terminology), so I end up with files that look like:

part-00001-{UUID}_00001.c000.snappy.parquet
part-00001-{UUID}_00002.c000.snappy.parquet
...
part-00001-{UUID}_00500.c000.snappy.parquet

part-00002-{UUID}_00001.c000.snappy.parquet
part-00002-{UUID}_00002.c000.snappy.parquet
...
part-00002-{UUID}_00500.c000.snappy.parquet

part-00500-{UUID}_00001.c000.snappy.parquet
part-00500-{UUID}_00002.c000.snappy.parquet
...
part-00500-{UUID}_00500.c000.snappy.parquet

That's 500x500=250000 bucketed parquet files! It takes forever for the FileOutputCommitter to commit that to S3.

Is there a way to generate one file per bucket, like in Hive? Or is there a better way to deal with this problem? As of now it seems like I have to choose between lowering the parallelism of my cluster (reduce number of writers) or reducing the parallelism of my parquet files (reduce number of buckets).

Thanks

2 Answers 2

6

In order to get 1 file per final bucket do the following. Right before writing the dataframe as table repartition it using exactly same columns as ones you are using for bucketing and set the number of new partitions to be equal to number of buckets you will use in bucketBy (or a smaller number which is a divisor of number of buckets, though I don't see a reason to use a smaller number here).

In your case that would probably look like this:

dataframe.repartition(500, bucketColumn1, bucketColumn2)
    .write()
    .format("parquet")
    .bucketBy(500, bucketColumn1, bucketColumn2)
    .mode(SaveMode.Overwrite)
    .option("path", "s3://my-bucket")
    .saveAsTable("my_table");

In the cases when you're saving to an existing table you need to make sure the types of columns are matching exactly (e.g. if your column X is INT in dataframe, but BIGINT in the table you're inserting into your repartitioning by X into 500 buckets won't match repartitioning by X treated as BIGINT and you'll end up with each of 500 executors writing 500 files again).

Just to be 100% clear - this repartitioning will add another step into your execution which is to gather the data for each bucket on 1 executor (so one full data reshuffle if the data was not partitioned same way before). I'm assuming that is exactly what you want.

It was also mentioned in comments to another answer that you'll need to be prepared for possible issues if your bucketing keys are skewed. It is true, but default Spark behavior doesn't exactly help you much if the first thing you do after loading the table is to aggregate/join on the same columns you bucketed by (which seems like a very possible scenario for someone who chose to bucket by these columns). Instead you will get a delayed issue and only see the skewness when try to load the data after the writing.

In my opinion it would be really nice if Spark offered a setting to always repartition your data before writing a bucketed table (especially when inserting into existing tables).

1
  • 1
    This should be accepted as the correct answer. It is unfortunate that Spark documentation makes this otherwise difficult to figure out since it seems it would be a common use case to want to re-bucket a table.
    – MikeGM
    May 26, 2021 at 18:58
-1

This should solve it.

dataframe.write()
  .format("parquet")
  .bucketBy(1, bucketColumn1, bucketColumn2)
  .mode(SaveMode.Overwrite)
  .option("path", "s3://my-bucket")
  .saveAsTable("my_table");

Modify the Input Parameter for the BucketBy Function to 1. You can look at the code of bucketBy from spark's git repository - https://github.com/apache/spark/blob/f8d59572b014e5254b0c574b26e101c2e4157bdd/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala

The first split part-00001, part-00002 is based on the number of parallel tasks running when you save the bucketed table. In your case you had 500 parallel tasks running. The number of files inside each part file is decided based on the input you provide for the bucketBy function.

To learn more about Spark tasks, partitions, executors, view my Medium articles - https://medium.com/@tharun026

3
  • 2
    This kind of make the bucketing pointless. I refer you to a thourough explanation about bucketing my colleagues have written - engineering.taboola.com/bucket-the-shuffle-out-of-here. It will generate less files per partition (up to 1 file per bucket per partition). If you need a single file per bucket you will need to shuffle by bucket keys, but that could result in severe skewness
    – Lior Chaga
    Jan 19, 2020 at 20:35
  • 4
    For give me for insisting, but seems this accepted answer is misleading. The whole idea of bucketing is to prepare the data for shuffle-less queries, assuming the data is bucketed according to some join key or aggregation keys. Using a single bucket is no different than just writing without any buckets.
    – Lior Chaga
    Jan 20, 2020 at 6:36
  • 1
    Not only this will likely create issues with using the data afterwards, this doesn't answer the question. The problem was that when using E executors and saving the data in a bucketed table with B buckets Spark generated E*B files (i.e. each of E executors splits its data into B buckets and writes B files kind of using file system as a way of repartitioning the data). When using one bucket as suggested here Spark will make each executor write one file, so the total number of files will be E - still E files per 1 bucket, not 1 file per bucket. May 24, 2020 at 23:49

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.