DISTINCT op 100 miljoen rijen forceert een volledige shuffle — en geen enkele Spark-configuratie elimineert dat

Een Databricks community thread legde een herkenbaar patroon bloot: een team dat 100 miljoen rijen logdata verwerkte over 1.000+ kolommen had al elke beschikbare optimalisatieknop ingeschakeld. AQE stond aan. Photon stond aan. Optimize Write en Auto Compaction waren beide actief. Ze hadden hun SELECT beperkt tot 20 kolommen vóór het aanroepen van .distinct(). Ze hadden spark.sql.shuffle.partitions verhoogd van de standaard 200 naar 400.

De job kroop nog steeds.

De Spark UI vertelde het verhaal duidelijk. De DISTINCT-fase domineerde de totale doorlooptijd en het shuffle write volume was groter dan alle andere fasen samen. Het team had alles gedaan wat de gangbare tuning-gidsen aanbevelen en liep toch tegen een muur.

Dit is het punt dat ook ervaren engineers kan verrassen: DISTINCT is geen filterbewerking. Het is een globale aggregatie. Spark moet elke rij hashen, rijen met dezelfde hash naar dezelfde executor sturen en ze daarna vergelijken. Met 100 miljoen rijen en 20 kolommen in de vergelijking moet elke individuele rij minimaal één keer over het netwerk. AQE kan lege partities samenvoegen na de shuffle en Photon kan het hashen versnellen, maar geen van beide kan de shuffle zelf overslaan. De data moet bewegen.

Een Databricks-medewerker bevestigde dit in de thread: exacte globale deduplicatie vereist inherent dat data wordt geshufeld. Het knelpunt is niet slechte optimalisatie. Het is de wiskundige realiteit van wat DISTINCT van Spark vraagt.

Wanneer Spark een DISTINCT uitvoert, kiest het één van twee physical operators: HashAggregate of SortAggregate. Het verschil is groter dan de meeste configuratieknopen.

HashAggregate bouwt een in-memory hash map van distinct keys. Snel, totdat de map het beschikbare geheugen overschrijdt en Spark naar schijf begint te schrijven. Op dat moment verandert een CPU-gebonden bewerking in een IO-gebonden bewerking en kan de fase-duur met een factor 5 tot 10 stijgen.

SortAggregate sorteert alle rijen op de dedup key en scant daarna lineair op duplicaten. Het verbruikt minder geheugen maar vereist altijd een volledige sortering, wat meer shuffle data en langere fasen oplevert, zelfs zonder spill.

Je kunt controleren welke operator Spark heeft gekozen door het physical plan te inspecteren:

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

Zoek in de output naar HashAggregate of SortAggregate. Als je SortAggregate ziet voor een job die in het geheugen zou moeten passen, heeft Spark besloten dat de key-breedte te groot is voor hashing. Met 20 string-kolommen is dat gebruikelijk. Sparks interne drempelwaarde is gebaseerd op de geschatte rij-grootte en brede string-keys sturen de aggregatie naar het sorteerpad.

De diagnostische stap die er het meest toe doet: controleer de spill metrics in de Spark UI. Navigeer naar de fase die de aggregatie uitvoert, klik op de taaksamenvatting en controleer "Shuffle Spill (Memory)" en "Shuffle Spill (Disk)". Als disk spill groter is dan nul, hebben je executors meer geheugen nodig of moet je verminderen wat er gehashed wordt. Partitie-tuning helpt niet als het echte probleem is dat elke executor zijn aandeel distinct keys niet in het geheugen past.

De meest effectieve oplossing die in de thread werd besproken, was geen configuratiewijziging. Het was een datatransformatie: vervang de vergelijking over 20 kolommen door één deterministische 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"])

In plaats van 20 kolommen per rij te shufflen, verstuurt Spark nu één string van 64 tekens. Het shuffle write volume daalt aanzienlijk. In het geval uit de thread rapporteerde het team een vermindering van ongeveer 60% aan shuffle bytes. Wat minstens zo belangrijk is: de smallere key houdt Spark op het HashAggregate-pad, omdat de geschatte rij-grootte voor de aggregatie-key comfortabel in de hash map past.

Twee kanttekeningen. Ten eerste is sha2 niet botsingsvrij. Bij 256 bits is de kans op een botsing over 100 miljoen rijen ongeveer 1 op 10^61, wat voor elke praktische pipeline effectief nul is. Ten tweede serialiseert to_json de struct deterministisch binnen één Spark-versie, maar de kolomvolgorde is belangrijk. Geef kolommen altijd in een expliciete, vaste volgorde mee aan struct() in plaats van een wildcard struct(*) te gebruiken die kan veranderen als het schema wijzigt.

Pre-repartitioneren op de hash key vóór de dedup biedt een verdere verbetering. Door df_with_hash.repartition(400, "row_fingerprint") aan te roepen vóór dropDuplicates, zorg je ervoor dat rijen met dezelfde fingerprint al co-located zijn. De daaropvolgende dedup wordt een lokale bewerking binnen elke partitie, waardoor de tweede shuffle die dropDuplicates anders zou triggeren, wordt geëlimineerd. De afweging: je shufflet éénmaal expliciet in plaats van Spark te laten beslissen, maar als je de verdeling van je data kent, is dit vrijwel altijd sneller dan AQE laten raden.

Volledige DISTINCT over de hele tabel bij elke pipeline run is het onderliggende antipatroon. Als je logdata dedupliceert die in dagelijkse of uurlijkse batches binnenkomt, hoef je de rijen van vandaag niet te vergelijken met elke rij die ooit is ingevoerd. Je hoeft ze alleen te vergelijken met een bekende set bestaande keys.

Delta Lake's MERGE-operatie pakt dit direct aan:

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 *

Dit patroon dedupliceert eerst binnen de batch (met de windowed ROW_NUMBER, die alleen de batch shufflet) en vergelijkt daarna met de doeltabel via een join op de fingerprint-kolom. Omdat Delta Z-ORDER- of OPTIMIZE-statistieken bijhoudt voor het doel, kan de join bestanden overslaan die geen overeenkomende keys bevatten. Het shuffle volume schaalt mee met de batchgrootte, niet met de totale tabelgrootte.

In het 100M-rijen scenario uit de community thread voerde het team ongeveer 2 miljoen nieuwe rijen per uur in. Door te schakelen van volledige DISTINCT naar incrementele MERGE daalde de dedup-fase van 45 minuten naar minder dan 3 minuten. Het verschil is niet subtiel. Het is het verschil tussen een job die in het schedulingvenster past en een job die dat niet doet.

Als je pipeline al Delta-tabellen gebruikt (en op Databricks zou dat zo moeten zijn), geeft de incrementele aanpak ook time-travel en audit-mogelijkheden. Je kunt DESCRIBE HISTORY target_table opvragen om precies te zien wanneer elke merge heeft plaatsgevonden en hoeveel rijen zijn ingevoegd. Dat is relevant wanneer een stakeholder vraagt waarom een getal is veranderd tussen gisteren en vandaag.

Zelfs met synthetische keys en incrementele patronen kunnen dedup jobs nog steeds ondermaats presteren. Dit zijn de specifieke diagnostieken die de werkelijke oorzaak aan het licht brengen in plaats van opnieuw te raden met Spark-configuraties.

1. Aantal shuffle partities vs. datavolume. De standaard spark.sql.shuffle.partitions is 200. Voor 100M rijen levert dat partities van ongeveer 500K rijen op. Prima voor smalle keys, te groot voor brede. Zet spark.sql.adaptive.coalescePartitions.initialPartitionNum op een hogere waarde (800–1000) en laat AQE omlaag samenvoegen. Zo geef je de optimizer ruimte zonder oversized partities af te dwingen.

2. Executor-geheugen vs. spill. Controleer spark.sql.adaptive.advisoryPartitionSizeInBytes (standaard 64MB). Als je shuffle partities deze waarde overschrijden na AQE-samenvoeging, verwerkt elke taak meer data dan bedoeld. Verlaag dit naar 32MB voor dedup workloads met hoge kardinaliteit.

3. Skewed keys. Als één fingerprint-waarde een onevenredig groot aandeel rijen vertegenwoordigt (gebruikelijk bij kolommen met veel nulls in logdata), raakt één executor overbelast terwijl de rest niets doet. De skew join optimalisatie van AQE helpt bij joins maar is niet van toepassing op aggregaties. Dit moet je handmatig afhandelen: filter bekende high-frequency duplicaten eruit vóór de dedup-fase.

4. Driver-side bottleneck. Als de Spark UI laat zien dat taken snel worden afgerond terwijl de fase-duur lang is, kan de driver taakresultaten sequentieel serialiseren. Verhoog spark.driver.maxResultSize boven de standaard 1GB als je SparkException: Total size of serialized results-fouten ziet.

5. File listing overhead. Voor zeer grote Delta-tabellen kan de initiële bestandslijsting vóór de merge minuten in beslag nemen. Voer regelmatig OPTIMIZE target_table ZORDER BY (row_fingerprint) uit en controleer of delta.dataSkippingNumIndexedCols je fingerprint-kolom omvat.

MetricSign bewaakt de duur van Databricks job-fasen en signaleert wanneer een dedup-fase verder afwijkt van de historische basislijn dan normaal, vóórdat de job zijn SLA mist. Als shuffle spill een fase van 3 minuten naar 25 minuten stuurt, groepeert het incident de regressie met de specifieke fase en clusterconfiguratie, zodat je de oorzaak naast de melding ziet in plaats van met een blinde zoektocht te moeten beginnen.

Pipeline-teams merken zelden dat dedup-prestaties verslechteren. De job die vorig kwartaal 12 minuten kostte, duurt nu 18 minuten. Niemand past het schedule aan. Dan piekt het datavolume (een nieuwe event-bron, een schema change die kolommen toevoegt) en mist de job zijn venster. Downstream Power BI datasets verversen op verouderde tabellen. Het financeteam opent een ticket.

Dit is een cumulatief probleem. Elke toegevoegde kolom vergroot de shuffle payload. Elke nieuwe databron verhoogt de kardinaliteit. De DISTINCT die werkte bij 50 miljoen rijen wordt onhoudbaar bij 200 miljoen en het failure mode is geleidelijke vertraging, niet een duidelijke fout.

De oplossing zit in de architectuur, niet in de configuratie. Behandel deduplicatie als een ontwerpbeslissing, niet als een standaardbewerking. Stel jezelf drie vragen voordat je dedup-logica schrijft: kun je een natural key definiëren? Zo ja, gebruik dropDuplicates op die key alleen. Kun je incrementeel verwerken? Zo ja, gebruik Delta MERGE. Heb je echt exacte globale dedup over de volledige geschiedenis nodig? Accepteer dan de shuffle-kosten, dimensioneer je cluster dienovereenkomstig en monitor de fase-duur als leading indicator van pipeline-gezondheid.

Teams die voorkomen dat ze om 2 uur 's nachts worden gewekt door dedup-failures zijn de teams die deze beslissingen expliciet hebben gemaakt tijdens het ontwerp, niet de teams die zich na het feit een weg uit een DISTINCT-knelpunt hebben getund.