Re: [External] Bucketing performance question

[Ofir Manor <om...@speedata.io>]

Try using spark.sql.legacy.bucketedTableScan.outputOrdering = true , which is supposed to tell Spark to assume buckets are pre-sorted
shuffle - Spark Sort Operation in Join Despite Pre-sorted Bucketed Tables - Stack Overflow<https://stackoverflow.com/questions/78442540/spark-sort-operation-in-join-despite-pre-sorted-bucketed-tables>
   Ofir
________________________________
From: Daniel Vol <dv...@undertone.com>
Sent: Tuesday, August 20, 2024 10:25 PM
To: user@spark.apache.org <us...@spark.apache.org>
Cc: aronovich.daniel@gmail.com <ar...@gmail.com>
Subject: [External] Bucketing performance question

Hello,

We have a great scenario for using bucketing (at least as per definition) :
Few data tables built hourly and partitioned by date + hour
Several processes that join those tables by ID field many times

One table is pretty big (~100G hourly), others are relatively small (but not small enough for broadcast), all joins are LEFT into the big table.

To test bucketing I took one day of data (~4.5TB in a big table) and only two tables - one big and one small, left join between and count as an action.

As a result on a cluster of 10 i4g.4xl (160 cores, 1T mem + ssd disks):
- non bucketed tables ~ 3 min to perform count
- bucketed to 160 + sort / sortWithinPartition ~ 5 min to perform count (huge mem + disk spill)
- bucketed to 16000 + sort ~ 4 min to perform count (no spill at all)

this is the data from spark ui:
non bucketed:
[image.png]
[image.png]
So we can see it reads 31525 files and 470G data (out of 4.5T total file size), uses shuffle (1000 partitions) and finishes in 3.1 min

Bucketed (to 16K buckets):
SortMerge JOIN:
[image.png]
[image.png]
Here we see it reads 16000 files and 245G of data (out of 4.5T total file size) NO shuffle, BUT SORT instead

Using hint for SHUFFLE_HASH Join improves a bit but still bad
t1.join(t2.hint("SHUFFLE_HASH"), col("id1") === col("id2"), "left).count

[image.png]

[image.png]


The code to create data:
df1
  .withColumn("id1", ...)
  .repartition(16000, col("id1"))
  .write
  .sortBy("id1") // tried with sortWithinPartition instead - faster write, same read
  .bucketBy(16000, "id1")
  .partitionBy("dt", "hr")
  .format("parquet").mode("overwrite")
  .option("path", "...")
  .saveAsTable("table_1_16000")

df2
  .withColumn("id2", ...)
  .repartition(16000, col("id2"))
  .write
  .sortBy("id2")
  .bucketBy(16000, "id2")
  .partitionBy("dt", "hr")
  .format("parquet").mode("overwrite")
  .option("path", "...")
  .saveAsTable("table_2_16000")

To test join (in Zeppelin):
val t1 = spark.sql("select * from table_1_16000 where dt = '...' ")
val t2 = spark.sql("select * from table_2_16000 where dt = '...' ")
t1.join(t2, col("id1") === col("id2"), "left).count

So it looks like bucketing doesn't use parquet metadata to check if data is already sorted on read - seriously affecting performance.

Just to mention - I tried to run it with 160 buckets (instead of 16000) and though each task reads only 1.4G of data => ~23G per instance (each instance has has 120G of memory + 3.5T dedicated ssd disk), I get a huge spill for both mem + disk - per my assumption due to heavy (likely unnecessary) sort:
[image.png]

Few questions:
- is Spark supposed to use sort information from a parquet file?
- if it is - should there be any difference in read after using sort or sortWithinPartition?
- is there a way somehow to force it to use sort metadata or skip sort during execution?

It is very odd that full shuffling performs much (25%) faster than bucketing.
Especially when with bucketing it reads half amount of files (16K instead of 35K) and much less total input data (270G instead of 470G)

Thanks!

Daniel