Re: Parquet 'bucketBy' creates a ton of files
It really depends on the use case. Bucketing is storing the data already
hash-partitioned. So, if you frequently perform aggregations or joins on the
bucketing column(s) then it can save you a shuffle. You need to keep in mind
that for joins to completely avoid a shuffle both tables would need to have the
same bucketing.
Sorting the data may help with filtering assuming you’re using a file format
like Parquet (e.g. if you frequently filter by account id). If you look at
slide 11 in this talk I gave at Summit you can see a simple example:
https://www.slideshare.net/databricks/lessons-from-the-field-episode-ii-applying-best-practices-to-your-apache-spark-applications-with-silvio-fiorito
From: Gourav Sengupta
Date: Wednesday, July 10, 2019 at 3:14 AM
To: Silvio Fiorito
Cc: Arwin Tio , "user@spark.apache.org"
Subject: Re: Parquet 'bucketBy' creates a ton of files
yeah makes sense, also is there any massive performance improvement using
bucketBy in comparison to sorting?
Regards,
Gourav
On Thu, Jul 4, 2019 at 1:34 PM Silvio Fiorito
mailto:silvio.fior...@granturing.com>> wrote:
You need to first repartition (at a minimum by bucketColumn1) since each task
will write out the buckets/files. If the bucket keys are distributed randomly
across the RDD partitions, then you will get multiple files per bucket.
From: Arwin Tio mailto:arwin@hotmail.com>>
Date: Thursday, July 4, 2019 at 3:22 AM
To: "user@spark.apache.org<mailto:user@spark.apache.org>"
mailto:user@spark.apache.org>>
Subject: Parquet 'bucketBy' creates a ton of files
I am trying to use Spark's **bucketBy** feature on a pretty large dataset.
```java
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-1-{UUID}_1.c000.snappy.parquet
part-1-{UUID}_2.c000.snappy.parquet
...
part-1-{UUID}_00500.c000.snappy.parquet
part-2-{UUID}_1.c000.snappy.parquet
part-2-{UUID}_2.c000.snappy.parquet
...
part-2-{UUID}_00500.c000.snappy.parquet
part-00500-{UUID}_1.c000.snappy.parquet
part-00500-{UUID}_2.c000.snappy.parquet
...
part-00500-{UUID}_00500.c000.snappy.parquet
```
That's 500x500=25 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), which
will lower the parallelism of my downstream jobs.
Thanks
Re: Parquet 'bucketBy' creates a ton of files
yeah makes sense, also is there any massive performance improvement using
bucketBy in comparison to sorting?
Regards,
Gourav
On Thu, Jul 4, 2019 at 1:34 PM Silvio Fiorito
wrote:
> You need to first repartition (at a minimum by bucketColumn1) since each
> task will write out the buckets/files. If the bucket keys are distributed
> randomly across the RDD partitions, then you will get multiple files per
> bucket.
>
>
>
> *From: *Arwin Tio
> *Date: *Thursday, July 4, 2019 at 3:22 AM
> *To: *"user@spark.apache.org"
> *Subject: *Parquet 'bucketBy' creates a ton of files
>
>
>
> I am trying to use Spark's **bucketBy** feature on a pretty large dataset.
>
>
>
> ```java
>
> 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-1-{UUID}_1.c000.snappy.parquet
>
> part-1-{UUID}_2.c000.snappy.parquet
>
> ...
>
> part-1-{UUID}_00500.c000.snappy.parquet
>
>
>
> part-2-{UUID}_1.c000.snappy.parquet
>
> part-2-{UUID}_2.c000.snappy.parquet
>
> ...
>
> part-2-{UUID}_00500.c000.snappy.parquet
>
>
>
> part-00500-{UUID}_1.c000.snappy.parquet
>
> part-00500-{UUID}_2.c000.snappy.parquet
>
> ...
>
> part-00500-{UUID}_00500.c000.snappy.parquet
>
> ```
>
>
>
> That's 500x500=25 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), which will lower the parallelism of my downstream jobs.
>
>
>
> Thanks
>
Re: Parquet 'bucketBy' creates a ton of files
You need to first repartition (at a minimum by bucketColumn1) since each task
will write out the buckets/files. If the bucket keys are distributed randomly
across the RDD partitions, then you will get multiple files per bucket.
From: Arwin Tio
Date: Thursday, July 4, 2019 at 3:22 AM
To: "user@spark.apache.org"
Subject: Parquet 'bucketBy' creates a ton of files
I am trying to use Spark's **bucketBy** feature on a pretty large dataset.
```java
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-1-{UUID}_1.c000.snappy.parquet
part-1-{UUID}_2.c000.snappy.parquet
...
part-1-{UUID}_00500.c000.snappy.parquet
part-2-{UUID}_1.c000.snappy.parquet
part-2-{UUID}_2.c000.snappy.parquet
...
part-2-{UUID}_00500.c000.snappy.parquet
part-00500-{UUID}_1.c000.snappy.parquet
part-00500-{UUID}_2.c000.snappy.parquet
...
part-00500-{UUID}_00500.c000.snappy.parquet
```
That's 500x500=25 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), which
will lower the parallelism of my downstream jobs.
Thanks
Re: Parquet 'bucketBy' creates a ton of files
Hi, Arwin.
If I understand you correctly, this is totally expected behaviour.
I don't know much about saving to S3 but maybe you could write to HDFS
first then copy everything to S3? I think the write to HDFS will probably
be much faster as Spark/HDFS will write locally or to a machine on the same
LAN. After writing to HDFS, you can then iterate over the resulting
sub-directories (representing each bucket) and coalesce the files in them.
Regards,
Phillip
On Thu, Jul 4, 2019 at 8:22 AM Arwin Tio wrote:
> I am trying to use Spark's **bucketBy** feature on a pretty large dataset.
>
> ```java
> 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-1-{UUID}_1.c000.snappy.parquet
> part-1-{UUID}_2.c000.snappy.parquet
> ...
> part-1-{UUID}_00500.c000.snappy.parquet
>
> part-2-{UUID}_1.c000.snappy.parquet
> part-2-{UUID}_2.c000.snappy.parquet
> ...
> part-2-{UUID}_00500.c000.snappy.parquet
>
> part-00500-{UUID}_1.c000.snappy.parquet
> part-00500-{UUID}_2.c000.snappy.parquet
> ...
> part-00500-{UUID}_00500.c000.snappy.parquet
> ```
>
> That's 500x500=25 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), which will lower the parallelism of my downstream jobs.
>
> Thanks
>
