[VL] Gluten write more shuffle data than vanilla Spark

[boneanxs]

Backend

VL (Velox)

Bug description

I can see that for the same record numbers, gluten always write much larger shuffle data than vanilla, why? From my understanding, gluten should use less storage since it's written in columnar format.

gluten

vanilla

Spark version

Spark-3.2.x

Spark configurations

No response

System information

gluten version: 1.3.0

Relevant logs

[marin-ma]

Could you share the data schema of the exchange? How many columns are there? We do observed larger compressed size and worse performance with growing number of columns. You can set spark.gluten.sql.columnar.shuffle.sort.columns.threshold to a smaller value to enable sort-based shuffle for this case.

[FelixYBW]

Can you try sort based shuffle?
dest batch is small.

[boneanxs]

The output consists of two columns, and the partition number is set to 100.

Input [3]: [hash_partition_key#1821, strategies#193, attributes#197]
Arguments: hashpartitioning(cast(get_json_object(attributes#197, $.SellerID) as bigint), 100, None), ENSURE_REQUIREMENTS, [strategies#193, attributes#197], [id=#1768], [shuffle_writer_type=hash]

by changing to sort type, it looks normal now.

Besides, I see these 2 values are set very high, do we want to avoid using sort based shuffle(maybe hash shuffle usually is faster)?

  val COLUMNAR_SHUFFLE_SORT_PARTITIONS_THRESHOLD =
    buildConf("spark.gluten.sql.columnar.shuffle.sort.partitions.threshold")
      .internal()
      .doc("The threshold to determine whether to use sort-based columnar shuffle. Sort-based " +
        "shuffle will be used if the number of partitions is greater than this threshold.")
      .intConf
      .createWithDefault(100000)

  val COLUMNAR_SHUFFLE_SORT_COLUMNS_THRESHOLD =
    buildConf("spark.gluten.sql.columnar.shuffle.sort.columns.threshold")
      .internal()
      .doc("The threshold to determine whether to use sort-based columnar shuffle. Sort-based " +
        "shuffle will be used if the number of columns is greater than this threshold.")
      .intConf
      .createWithDefault(100000)

Are there any best practices for selecting between hash-based and sort-based shuffle when using Gluten? If we can automate this selection, it could simplify future upgrades to Gluten. Now some jobs get poor performance because of this.

[marin-ma]

@boneanxs Thanks for the update. Sort-based shuffle is a new feature added in recent gluten release, and it's disabled by default unless explicitly configured by users.

We have benchmarked the performance of hash vs. sort shuffle based on evenly partitioned data and observed that sort-based shuffle only improves performance when the number of columns is greater than 8 or the number of partitions exceeds 8K. However, it seems that the case you provided involves data skew. Therefore, our benchmark values do not apply to your case (only 2 columns and 100 partitions). Perhaps we need to conduct some tests for skewed data and update the documentation with the suggested values. cc: @FelixYBW @zhouyuan

Note that the configuration will enable sort-based shuffle across the entire job. Currently, it cannot be controlled for individual shuffle.

[boneanxs]

@marin-ma Thanks for your detailed reply! Maybe we can dynamically change the shuffle type leveraged by AQE if skewed data is also a factor

[marin-ma]

@boneanxs AQE depends on the analysis of the output from shuffle write. It seems that we cannot leverage AQE to help choose the shuffle writer type.

[boneanxs]

@marin-ma @FelixYBW I noticed that even after updating some jobs to use sort-based shuffle, the shuffle data is still much larger compared to vanilla Spark. what could be the cause?

Gluten:

Vanilla:

[boneanxs]

Between

Note that the configuration will enable sort-based shuffle across the entire job. Currently, it cannot be controlled for individual shuffle.

I think it can control individual shuffle, since the value is calculated inside ColumnarShuffleExchangeExec, and each shuffle has its own shuffleWriterType, the different shuffle should use different shuffleWriterType depending on its partition size and column size?

[FelixYBW]

@marin-ma @FelixYBW I noticed that even after updating some jobs to use sort-based shuffle, the shuffle data is still much larger compared to vanilla Spark. what could be the cause?

Looks they are simple schema. Shuffle size is 70G vs. 109G

[marin-ma]

@boneanxs
Are you still using Gluten 1.3 for the latest test? There was an optimization for the compression size with sort based shuffle. It's will be included in Gluten 1.5 release. #9278

[boneanxs]

@boneanxs Are you still using Gluten 1.3 for the latest test? There was an optimization for the compression size with sort based shuffle. It's will be included in Gluten 1.5 release. #9278

Yes, we're still using 1.3 as our internal version, will have a try this pr. Thanks!