DISTINCT is the major bottleneck because of the heavy shuffle.

[Nidhig631]

Need some advice from the community.

I am processing around 100 million records using:

df.select(required_cols).distinct().write.saveAsTable(...)

The source has 1000+ columns, but I'm selecting only 20 columns before applying DISTINCT.

I have already enabled:

• AQE
• Photon
• Optimize Write
• Auto Compaction
• Shuffle Partition to 400
• coalescePartitions

DISTINCT is still the biggest bottleneck due to the shuffle.

For log data that cannot rely on any primary key or composite key, what is the best approach for deduplication at scale?
Worker Type is Standard_DS15_v2, and min:5 to max:7 workers is the current configuration.

[bala_sai]

I think a hash based approach is worth trying here. Since you are deduping on 20 columns, distinct() has to shuffle and compare all those columns, which can become expensive at this scale.

With a hash column, Spark still has to shuffle, but the comparison becomes lighter because it is working with one deterministic hash value instead of all 20 columns. So it may not remove the bottleneck completely, but it can reduce the shuffle/comparison cost.

There will be some extra CPU cost to generate the hash, so I would still benchmark both approaches. For recurring log dedup, I would lean toward generating a hash from the selected fields and using it as a technical dedup key, especially if the process can be made incremental instead of deduping the full dataset every run.

[Nidhig631]

The data is overwrite mode anyways, just one day data.Distinct working is same as generating hashing key at the background for my use case.I have already checked on the solution suggested by you.

[balajij8]

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

[Nidhig631]

I have already checked spark UI there is no data skewness, for such data load shuffle partition 400 is best.The third suggestion I will check but cost may increase in this.

[Ashwin_DSA]

Hi @Nidhig631,

DISTINCT is still an exact global deduplication step, and that means Spark has to shuffle rows so identical values can meet in the same place. So, what you are seeing is normal. Selecting 20 columns instead of 1000 definitely reduces the amount of data being shuffled, but it doesn’t eliminate the fundamental cost of the operation.

For this kind of log workload, I would usually avoid thinking in terms of "how do I make distinct() faster?" and instead focus on "how do I make the dedup key narrower and more reusable?" If those 20 columns are what define a duplicate, a better pattern is to build a deterministic fingerprint from them like @bala_sai has mentioned... for example, with sha2(to_json(struct(...)), 256), and then de-duplicate on that fingerprint with dropDuplicates(). That still requires a shuffle for exact deduplication, but you are now shuffling and comparing a much narrower key instead of the full 20-column payload, and you also get a synthetic key you can persist for future incremental loads. Databricks documentation also recommends using dropDuplicates() or dropDuplicatesWithinWatermark() rather than relying on distinct() when the goal is deduplication on specific fields.

I’d be a little careful with the idea that repartitioning by a hash of all 20 columns makes the rest of the dedup "local" and therefore removes the shuffle cost. It can help colocate likely duplicates, but it does not automatically guarantee Spark won’t introduce another exchange later. The safe way to treat that optimisation is as "potentially useful, but something to verify with explain("formatted")," not as a guaranteed shuffle elimination.

On the tuning side, increasing shuffle parallelism is probably worth testing, but I would not jump straight to a fixed number like 800 or 1600 as a universal answer. Databricks recommends using Adaptive Query Execution and auto-optimized shuffle, because the right shuffle width depends on the actual data size and the reducer stage, not just cluster size. If you are currently fixed at 400, trying spark.sql.shuffle.partitions=auto is a very reasonable next step.

Persisting the 20-column projection can help if you reuse it multiple times, but by itself it usually does not change the cost profile of the dedup. If the narrowed DataFrame is only used once, the bigger lever is still the dedup strategy itself rather than caching or checkpointing.

If this is a recurring ingestion pattern rather than a one-time cleanup, the best long-term approach is usually incremental deduplication. In other words, compute the synthetic dedup key once, store it in Delta, and for new data only compare incoming keys against previously accepted keys instead of re-running a full-table exact dedup every time. That tends to be a much bigger win than trying to squeeze a little more out of a global DISTINCT.

If the data is actually time-bounded and duplicates only matter within a lateness window, then a watermark-based approach is even better, and Databricks has specific guidance for that in the Structured Streaming deduplication docs.

Hope this helps.

If this answer resolves your question, could you mark it as “Accept as Solution”? That helps other users quickly find the correct fix.

[Nidhig631]

Thanks for your response. However, I found it to be quite generic and more conversational than solution-oriented. I was expecting recommendations tailored to my specific problem statement, along with actionable steps or technical approaches to address the issue. Could you please provide a more focused solution based on the details shared?

[Ashwin_DSA]

So, based on my first recommendation, here is a sample pattern.

from pyspark.sql import functions as F

cols = required_cols

dedup_df = (
    df.select(*cols)
      .withColumn(
          "dedup_key",
          F.sha2(F.to_json(F.struct(*[F.col(c) for c in cols])), 256)
      )
      .repartition("dedup_key")
      .dropDuplicates(["dedup_key"])
      .drop("dedup_key")
)

dedup_df.write.mode("overwrite").saveAsTable("target_table")

The reason I prefer this over raw distinct() is not that it eliminates the shuffle completely, because it does not. The benefit is that Spark now shuffles and compares a narrow synthetic key instead of carrying the full 20-column row through the dedup logic, and that same key can later be reused if you move to incremental dedup. However, I also note your comment that you have already tried this and have not seen any improvement. Is my understanding correct? 

Since you have already checked the Spark UI and confirmed there is no meaningful skew, and since you have already tested different partition counts and found 400 to be the best setting for this load, I would not keep pushing on shuffle tuning. That tells us the bottleneck is not poor distribution or under-partitioning anymore. At that point, the remaining cost is the exact deduplication itself, which is expected for a global DISTINCT over 100M rows.

I would still check the physical plan once with explain("formatted"), but only for one specific reason...to confirm whether Spark is using a HashAggregate-based dedup path or falling back to SortAggregate. If it is SortAggregate, that can make the query materially more expensive because Spark has to sort within partitions before aggregating. If it is already using HashAggregate, then that confirms the main issue is simply the unavoidable cost of exact global dedup, not a bad execution strategy.

[Yogasathyandrun]

I think you’re looking at this the right way.

For an exact dedup, there’s no way around a global shuffle somewhere. A separate hash column doesn’t fundamentally change that, since Spark is already hashing the grouping keys internally for distinct().

Since you’ve already ruled out skew, I’d spend more time looking at spill than at the dedup logic itself. In my experience, once skew is gone, the next question is whether the shuffle is memory-bound or simply moving a lot of data.

One thing I’d also validate is where the runtime is actually being spent. I’ve seen cases where people attribute everything to the distinct(), but a noticeable chunk of the wall-clock time is in the write path afterwards, especially when Optimize Write and Auto Compaction are both enabled.

If the UI shows balanced tasks, minimal spill, and no skew, you may already be fairly close to the practical limit for an exact global dedup of this size.

[Nidhig631]

From my analysis, the write phase is completing relatively quickly. The primary performance issue seems to be the DISTINCT operation, which is taking a significant amount of time due to the shuffle and deduplication overhead.

[Yogasathyandrun]

That’s useful context.

If skew is already ruled out and the write phase is relatively fast, I’d focus on two Spark UI metrics next: shuffle write volume and spill (memory/disk). Those will tell you whether there’s still a tuning opportunity or whether you’re simply paying the unavoidable cost of a large global dedup.

One thing I’d be curious about: do duplicates tend to arrive close together in the source data (for example, log retransmits or replay scenarios)? If so, improving duplicate locality before the shuffle can sometimes reduce the amount of data that survives into the global dedup stage.