DISTINCT on 100M Rows Forces a Full Shuffle — and No Spark Config Can Eliminate It

A Databricks community thread captured a familiar pattern: a team processing 100 million rows of log data across 1,000+ columns had already enabled every optimization toggle available. AQE was on. Photon was on. Optimize Write and Auto Compaction were both enabled. They'd narrowed their SELECT to 20 columns before calling .distinct(). They'd bumped spark.sql.shuffle.partitions from the default 200 to 400.

The job still crawled.

The Spark UI told the story clearly. The DISTINCT stage dominated wall-clock time, and the shuffle write volume dwarfed every other stage combined. The team had done everything the standard tuning guides recommend — and hit a wall.

This is the part that trips up experienced engineers: DISTINCT is not a filtering operation. It is a global aggregation. Spark must hash every row, redistribute rows with the same hash to the same executor, then compare them. If you have 100 million rows and 20 columns in the comparison, every single row must cross the network at least once. AQE can coalesce empty partitions after the shuffle, and Photon can speed up the hashing, but neither can skip the shuffle itself. The data must move.

A Databricks employee confirmed this in the thread: exact global deduplication inherently requires shuffling data. The bottleneck isn't poor optimization. It's the mathematical reality of what DISTINCT asks Spark to do.

When Spark executes a DISTINCT, it picks one of two physical operators: HashAggregate or SortAggregate. The difference matters more than most config knobs.

HashAggregate builds an in-memory hash map of distinct keys. It is fast — until the map exceeds available memory and Spark starts spilling to disk. At that point, what was a CPU-bound operation becomes IO-bound, and your stage time can increase by 5–10x.

SortAggregate sorts all rows by the dedup key, then scans linearly for duplicates. It uses less memory but always requires a full sort, which means more shuffle data and longer stage times even without spill.

You can check which operator Spark chose by inspecting the physical plan:

df.select(required_cols).distinct().explain(True)

Look for HashAggregate or SortAggregate in the output. If you see SortAggregate on a job that should fit in memory, Spark has decided the key width is too large for hashing. With 20 string columns, this is common. Spark's internal threshold is based on estimated row size, and wide string keys push aggregation toward the sort path.

The diagnostic step that matters most is checking spill metrics in the Spark UI. Navigate to the stage performing the aggregate, click into the task summary, and check "Shuffle Spill (Memory)" and "Shuffle Spill (Disk)." If disk spill exceeds zero, your executors need more memory or you need to reduce what's being hashed. No amount of partition tuning helps when the fundamental issue is that each executor can't hold its share of distinct keys in memory.

The most effective fix discussed in the thread was not a configuration change. It was a data transformation: replace the 20-column comparison with a single deterministic hash.

from pyspark.sql.functions import sha2, to_json, struct

df_with_hash = df.withColumn(
    "row_fingerprint",
    sha2(to_json(struct(*required_cols)), 256)
)

df_deduped = df_with_hash.dropDuplicates(["row_fingerprint"])

Instead of shuffling 20 columns per row, Spark now shuffles one 64-character string. The shuffle write volume drops dramatically — in the thread's case, the team reported a roughly 60% reduction in shuffle bytes. More importantly, the narrower key keeps Spark on the HashAggregate path because the estimated row size for the aggregate key fits comfortably in the hash map.

Two caveats. First, sha2 is not collision-free. At 256 bits, the probability of a collision across 100 million rows is approximately 1 in 10^61, which is effectively zero for any practical pipeline. Second, to_json serializes the struct deterministically within a single Spark version, but column ordering matters. Always pass columns to struct() in an explicit, fixed order rather than using struct(*) with a wildcard that could change if the schema evolves.

Pre-repartitioning on the hash key before dedup offers a further improvement. By calling df_with_hash.repartition(400, "row_fingerprint") before dropDuplicates, you ensure that rows with the same fingerprint are already co-located. The subsequent dedup becomes a local operation within each partition, eliminating the second shuffle that dropDuplicates would otherwise trigger. The tradeoff is that you shuffle once explicitly instead of letting Spark decide, but when you know your data's distribution, this is almost always faster than letting AQE guess.

Full-table DISTINCT on every pipeline run is the root cause pattern. If you're deduplicating log data that arrives in daily or hourly batches, you don't need to compare today's rows against every row ever ingested. You need to compare them against a known set of existing keys.

Delta Lake's MERGE operation handles this directly:

MERGE INTO target_table t
USING (
  SELECT * FROM new_batch
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY row_fingerprint ORDER BY event_time DESC
  ) = 1
) s
ON t.row_fingerprint = s.row_fingerprint
WHEN NOT MATCHED THEN INSERT *

This pattern deduplicates within the batch first (using the windowed ROW_NUMBER, which only shuffles the batch), then checks against the target table using a join on the fingerprint column. Because Delta maintains Z-ORDER or OPTIMIZE statistics on the target, the join can skip files that don't contain matching keys. The shuffle volume scales with batch size, not total table size.

For the 100M-row scenario in the community thread, the team was ingesting roughly 2 million new rows per hour. Switching from full-table DISTINCT to incremental MERGE reduced the dedup stage from 45 minutes to under 3 minutes. The difference isn't subtle. It's the difference between a job that fits in a scheduling window and one that doesn't.

If your pipeline already uses Delta tables (and on Databricks, it should), the incremental approach also gives you time-travel and audit capability. You can query DESCRIBE HISTORY target_table to see exactly when each merge occurred and how many rows were inserted, which matters when a stakeholder asks why a number changed between yesterday and today.

Even with synthetic keys and incremental patterns, dedup jobs can still underperform. These are the specific diagnostics that surface the actual cause rather than prompting another round of config guessing.

1. Shuffle partition count vs. data volume. The default spark.sql.shuffle.partitions is 200. For 100M rows, this creates partitions of roughly 500K rows each — fine for narrow keys, too large for wide ones. Set spark.sql.adaptive.coalescePartitions.initialPartitionNum to a higher value (800–1000) and let AQE coalesce downward. This gives the optimizer room to work without forcing oversized partitions.

2. Executor memory vs. spill. Check spark.sql.adaptive.advisoryPartitionSizeInBytes (default 64MB). If your shuffle partitions exceed this after AQE coalescing, each task processes more data than intended. Lower it to 32MB for high-cardinality dedup workloads.

3. Skewed keys. If one fingerprint value accounts for a disproportionate share of rows (common with null-heavy log columns), a single executor gets buried while others idle. AQE's skew join optimization helps for joins but does not apply to aggregations. You must handle this manually — filter out known high-frequency duplicates before the dedup stage.

4. Driver-side bottleneck. If the Spark UI shows tasks completing quickly but stage time is long, the driver may be serializing task results sequentially. Increase spark.driver.maxResultSize from the default 1GB if you see SparkException: Total size of serialized results errors.

5. File listing overhead. For very large Delta tables, the initial file listing before the merge can take minutes. Run OPTIMIZE target_table ZORDER BY (row_fingerprint) regularly, and confirm that delta.dataSkippingNumIndexedCols includes your fingerprint column.

MetricSign monitors Databricks job durations and surfaces when a dedup stage regresses beyond its historical baseline — before the job breaches its SLA. When shuffle spill causes a 3-minute stage to become a 25-minute stage, the incident groups the regression with the specific stage and cluster configuration, so you see the cause alongside the alert rather than starting a blind investigation.

Pipeline teams rarely notice dedup performance degrading. The job that took 12 minutes last quarter now takes 18 minutes. Nobody adjusts the schedule. Then data volume spikes — a new event source, a schema change that adds columns — and the job misses its window. Downstream Power BI datasets refresh against stale tables. The finance team opens a ticket.

This is a compounding problem. Each added column widens the shuffle payload. Each new data source increases cardinality. The DISTINCT that worked at 50 million rows becomes untenable at 200 million, and the failure mode is gradual slowdown, not a clean error.

The fix is architectural, not configurational. Treat deduplication as a design decision, not a default operation. Ask three questions before writing any dedup logic: Can you define a natural key? If yes, use dropDuplicates on that key alone. Can you process incrementally? If yes, use Delta MERGE. Do you genuinely need exact global dedup across the full history? If so, accept the shuffle cost, size your cluster accordingly, and monitor the stage duration as a leading indicator of pipeline health.

The teams that avoid 2am pages from dedup failures are the ones that made these decisions explicitly at design time — not the ones that tuned their way out of a DISTINCT bottleneck after it already broke.