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.