Databricks job failures: OOM, gegevensscheefheid en DRIVER_NOT_RESPONDING

Databricks job failures komen vaak voor bij een klein aantal hoofdoorzaken. Het gaat hierbij om geheugenproblemen (out of memory) op de driver of executor, data-scheefheid waardoor één partitie de geheugenlimiet van een executor ver overschrijdt, shuffle-fouten wanneer opgeslagen data na een node-uitval niet meer kan worden teruggelezen en driver-heartbeat-time-outs veroorzaakt door garbage collection-pauzes die lang genoeg duren om de clustercommunicatie te verbreken.

Deze patronen zijn aan elkaar gerelateerd. Data-scheefheid veroorzaakt een out of memory-fout op de overbelaste executor. Grote shuffles schrijven data naar de schijf en mislukken vervolgens wanneer een node opnieuw opstart. GC-pauzes worden veroorzaakt door druk op de heap, die ontstaat door het verzamelen van grote datasets naar de driver of door de accumulatie van shuffle-status.

Het identificeren van het specifieke patroon is bepalend voor de oplossing. Een out of memory-fout op de driver wordt anders opgelost dan een out of memory-fout op een executor. Shuffle-fouten vereisen een verhoging van de shuffle-partities, niet van het drivergeheugen. Snel de juiste diagnose stellen – zonder de hele job opnieuw uit te voeren – is cruciaal om te bepalen of een productiefout een incident van dertig minuten of van drie uur is.

DRIVER_NOT_RESPONDING is het meest zichtbare symptoom van een OutOfMemoryError (OOM) van de driver. De Spark-driver crasht omdat de heap vol raakt, door een collect()-aanroep die te veel data van de executors heeft opgehaald, door een grote broadcast join die de driver in het geheugen heeft klaargezet of door geaccumuleerde objectreferenties tijdens een langlopende job.

De eerste oplossing is het verhogen van het drivergeheugen in de configuratie van het jobcluster: spark.driver.memory van de standaardwaarde (vaak 4 GB) naar 8 GB of 16 GB, afhankelijk van de workload. Dit gebeurt in de compute-instellingen van de job, niet in de notebookcode, omdat het drivergeheugen moet worden ingesteld voordat de SparkSession start.

De tweede oplossing is architectonisch. Als de driver een OOM-fout veroorzaakt door collect()-aanroepen op grote DataFrames, moet het patroon veranderen, schrijf de resultaten naar Delta in plaats van ze naar de driver te verzamelen. Als grote broadcasts de oorzaak zijn, stel dan spark.sql.autoBroadcastJoinThreshold in op -1 om automatische broadcast joins uit te schakelen en Spark in plaats daarvan een sort-merge join te laten gebruiken.

Bij een 'executor OOM' (Out of Memory) wordt in de joblogboeken de foutmelding 'ExecutorLostFailure' weergegeven. Het vergroten van het geheugen van de executor of het verminderen van het aantal cores per executor (zodat elke executor minder partities tegelijk verwerkt) lost de meeste problemen op.

Data-scheefheid treedt op wanneer records na een shuffle ongelijk verdeeld zijn over partities. Eén partitie bevat tien miljoen rijen, terwijl de andere er vijftigduizend bevatten. De executor die de grote partitie verwerkt, draait twintig minuten, terwijl de andere executors binnen dertig seconden klaar zijn. De job wacht op de trage executor, loopt uiteindelijk vast door een time-out of de executor raakt door de belasting zonder geheugen.

Het diagnostische signaal is zichtbaar in de Spark UI: bekijk de verdeling van de jobduur voor de falende fase. Als één job aanzienlijk langer duurt dan alle andere, is scheefheid de oorzaak.

De oplossing is salting: voeg een willekeurig geheel getal toe aan de join-sleutel om records kunstmatig over meer partities te verdelen.

from pyspark.sql.functions import rand, col df_salted = df.withColumn("salt", (rand() * 10).cast("int")) df_joined = df_salted.join(other_df, ["key", "salt"])

De salt-waarde (10 in dit voorbeeld) bepaalt in hoeveel subpartities elke sleutelwaarde wordt opgesplitst. Grotere salt-waarden verminderen de scheefheid agressiever, maar verhogen het totale aantal shuffle-partities. Na het toevoegen van salt wordt de langzame partitie opgesplitst in tien ongeveer gelijke partities, waardoor de job proportioneel sneller wordt voltooid.

Shuffle-intensieve bewerkingen, joins, aggregaties, groeperen op grote datasets, vereisen dat Spark de data herverdeelt over de executors. De tussentijdse shuffle-bestanden worden op elke executor naar de schijf geschreven. Als een executor halverwege het shuffle-proces opnieuw opstart, worden de bestanden op die node onleesbaar voor executors die ze proberen op te halen en mislukt de fase met een shuffle fetch exception.

De configuratieoplossing voor de geheugendruk tijdens het shuffle-proces is het verhogen van spark.sql.shuffle.partitions boven de standaardwaarde van 200. Voor datasets van honderden gigabytes verkleinen waarden van 400 tot 1000 de grootte per partitie en verminderen ze het benodigde geheugen per executor tijdens de shuffle-fase.

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

Dit is een afweging: meer partities betekenen meer taken, wat extra overhead voor de planning met zich meebrengt. Maar voor grote datasets is de lagere geheugenbelasting per partitie de extra overhead vrijwel altijd waard.

Bij aanhoudende shuffle-fouten na herstarts van knooppunten, zorgt het inschakelen van speculatie (spark.speculation=true) ervoor dat Spark dubbele taken start voor trage taken en de job gebruikt die als eerste is voltooid, waardoor de kans op instabiliteit van knooppunten wordt verkleind.

De driver onderhoudt een heartbeat met de clustermanager met een vast interval (standaard: 60 seconden). Als de JVM van de driver langer dan dit interval gepauzeerd is voor een volledige garbage collection-cyclus, beschouwt de clustermanager de driver als dood en markeert de job als mislukt met DRIVER_NOT_RESPONDING of vergelijkbare time-outfouten.

De duur van de GC-pauze is evenredig met de bezetting van de heap. Hoe meer objecten er in de heap van de driver staan, hoe langer een volledige garbage collection duurt. De hoofdoorzaak is altijd heapdruk – ofwel door een workload die verdeeld had moeten worden over de executors, ofwel door een geheugenlek in de gebruikerscode (geaccumuleerde broadcastvariabelen, niet-gesloten accumulators, groeiende objectgrafieken in lussen).

GC-druk detecteren: open in de Spark UI het tabblad Executors of Driver en bekijk de GC-tijd als percentage van de jobtijd. Meer dan 10% van de tijd besteed aan GC is een signaal van geheugendruk. Boven de 20% is de drempel waarbij GC-pauzes heartbeat-time-outs kunnen veroorzaken.

Oplossingen: verhoog spark.driver.memoryOverhead (het deel van het geheugen dat niet op de heap wordt opgeslagen), schakel over naar G1GC als het cluster de standaard parallelle collector gebruikt en vermijd patronen die objecten in de driver accumuleren, met name collect()-aanroepen op resultaten die worden geïtereerd in plaats van weggeschreven.

Wanneer een Databricks-job halverwege mislukt, starten de meeste engineers de hele job opnieuw. Dit is meestal verkeerd: het verspilt tijd en rekenkracht aan het opnieuw uitvoeren van taken die al succesvol zijn afgerond.

De functie 'Repair Run' voert alleen de mislukte en overgeslagen taken van een eerdere jobuitvoering opnieuw uit, vanaf het punt waar de job is mislukt. Succesvolle taken worden niet opnieuw uitgevoerd. In een job met tien taken, waarbij job zeven is mislukt, voert 'Repair Run' de taken zeven tot en met tien opnieuw uit en hergebruikt de uitvoer van de taken één tot en met zes.

U kunt deze functie openen via de Lakeflow Jobs UI: zoek de mislukte uitvoering, klik op de uitvoerings-ID en selecteer 'Repair Run'. U kunt ook de Jobs REST API gebruiken:

POST /api/2.1/jobs/runs/repair {"run_id": , "rerun_tasks": ["task_key_7", "task_key_8", "task_key_9", "task_key_10"]}

De functie 'Repair Run' is vooral waardevol voor taken met kostbare upstream-taken, zoals het importeren van grote hoeveelheden data, het trainen van modellen of zware Delta-merge-bewerkingen. Het opnieuw uitvoeren van deze taken vanaf nul, wanneer alleen een downstream-job is mislukt, is kostbaar en onnodig.

Voor tijdelijke fouten, zoals netwerkproblemen of het overnemen van Azure Spot-instanties, kunt u herhalingsbeleid op jobniveau configureren (Maximum aantal herhalingen: 1-2, Opnieuw proberen bij time-out: true), zodat de job deze automatisch afhandelt zonder handmatige tussenkomst.

Universele clusters zijn geoptimaliseerd voor interactief gebruik. Ze starten eenmalig en blijven continu draaien. Ze schalen niet automatisch voor batchworkloads en ze slaan de JVM-status op tussen de uitvoeringen. Het gebruik van een universeel cluster voor een geplande productiejob betekent dat elke uitvoering de geheugenstatus van eerdere interactieve sessies overneemt – een betrouwbare bron van moeilijk te reproduceren OOM-fouten (Out of Memory).

Taakclusters zijn de juiste keuze voor productietaken. Ze starten elke keer opnieuw, gebruiken alleen de resources die de job nodig heeft, schakelen uit wanneer de uitvoering is voltooid en isoleren de uitvoeringen van elkaar. Ze zijn ook goedkoper: u betaalt alleen voor de duur van de jobuitvoering, niet voor de inactiviteit tussen de uitvoeringen.

Voor taken die toegang nodig hebben tot systeemtabellen voor monitoring, biedt de volgende query:

SELECT job_id, run_id, result_state, start_time, end_time, state_message FROM system.lakeflow.job_run_timeline WHERE result_state = 'FAILED' AND start_time >= DATEADD(day, -7, CURRENT_TIMESTAMP()) ORDER BY start_time DESC

een zeven dagen durende foutenhistorie rechtstreeks vanuit Databricks, inclusief foutmeldingen per uitvoering.

MetricSign bewaakt Databricks-taken via de Jobs REST API en genereert twee soorten incidenten: job_failed wanneer een uitvoering eindigt in een mislukte status (inclusief het exacte state_message van de uitvoering), en job_slow wanneer een uitvoering de historische P50-duur aanzienlijk overschrijdt, gedetecteerd met behulp van MAD-gebaseerde anomaliedetectie.

Het job_slow-signaal treedt eerder op. Een job die normaal gesproken binnen twaalf minuten wordt voltooid, maar nu vijfenveertig minuten duurt, is mogelijk nog niet mislukt, maar de kans is groot dat dit wel het geval zal zijn. De oorzaak is vaak een groeiend datavolume dat het geheugengebruik naar een kritiek punt duwt of een scheefheidspatroon in de data dat is verergerd naarmate de brontabel groter werd.

Door beide signalen te hebben, kunt u actie ondernemen vóór de storing in plaats van erna: onderzoek de trage job, vind de scheefheid of geheugendruk en los deze op voordat de volgende uitvoering een OOM-fout veroorzaakt.