You can follow below

1. Pre-partition by hash before DISTINCT - It ensures records with identical values land on the same partition upfront making the subsequent distinct operation local

from pyspark.sql.functions import hash, col

df.select(cols) \
  .repartition(400, *cols) \
  .distinct() \
  .write.saveAsTable()

2. Increase Shuffle Partitions - You can increase shuffle partitions to 800

spark.conf.set("spark.sql.shuffle.partitions", "800")

3. Worker configuration - You can use Standard_E16as_v4 workers instead of Standard_DS15_v2 for shuffle heavy workloads and min:6 to max:10 workers