Spark Optimalisatie in Fabric Notebooks: Gids voor Prestatieafstemming

De meeste mensen schrijven Spark notebooks en hopen dat ze snel draaien. Ze zijn dan verrast wanneer het verwerken van 10GB 20 minuten duurt of veel meer capacity kost dan verwacht.

Het probleem is dat men Spark behandelt als gewone Python. Dat is het niet. Je code is slechts instructies. Het echte werk gebeurt in een gedistribueerd systeem dat je correct moet configureren.

Er is een scheiding tussen logica en fysica. Logica is je code, welke transformaties je wilt. Fysica is de Spark cluster, hoe die transformaties daadwerkelijk worden uitgevoerd. Je moet beide begrijpen.

Het mentale model van logica vs. fysica

Logica: de transformaties die je schrijft in je notebook

df = spark.read.parquet("files/raw_data")
df = df.filter(df.date >= "2024-01-01")
df = df.groupBy("customer_id").agg(sum("amount"))
df.write.saveAsTable("aggregated_sales")

Fysica: de cluster die die code uitvoert

• Hoeveel nodes
• Hoeveel geheugen per node
• Hoeveel cores
• Netwerk tussen nodes
• Opslag I/O limieten

Je code is misschien perfect, maar als de fysica verkeerd is, zal het nog steeds traag zijn. Of je cluster is krachtig, maar inefficiënte code verspilt die kracht.

De meeste prestatieproblemen zijn fysica-gerelateerd, niet logica-gerelateerd. Mensen optimaliseren de verkeerde laag.

Fabric capacity SKUs: wat je daadwerkelijk krijgt

Fabric laat je clusters niet node voor node configureren zoals Databricks. In plaats daarvan krijg je capacity units op basis van je SKU, en Fabric beheert de Spark resources.

Dit is wat elke SKU je geeft voor Spark workloads:

F2 (trial/dev)

• 2 capacity units
• Effectief 1-2 Spark executors
• ~8GB bruikbaar geheugen in totaal
• Goed voor: leren, kleine datasets onder 1GB

F4

• 4 capacity units
• 2-4 executors
• ~16GB geheugen
• Goed voor: kleine productieworkloads, datasets 1-5GB

F8

• 8 capacity units
• 4-6 executors
• ~32GB geheugen
• Goed voor: de meeste standaard workloads, datasets 5-20GB

F16

• 16 capacity units
• 8-12 executors
• ~64GB geheugen
• Goed voor: grotere productieworkloads, datasets 20-50GB

F32

• 32 capacity units
• 16-24 executors
• ~128GB geheugen
• Goed voor: zware verwerking, datasets 50-200GB

F64 en hoger

• 64+ capacity units
• 32+ executors
• 256GB+ geheugen
• Goed voor: zeer grote schaal, datasets 200GB+

Dit zijn schattingen. Fabric toont geen exacte aantallen executors en deze variëren op basis van het workload-type. Maar dit geeft je een mentaal model van welke schaal elke tier aankan.

Kleine datasets: vermijd Spark volledig

Als je data minder dan 1GB is, gebruik dan helemaal geen Spark DataFrames. Gebruik Pandas.

Spark heeft overhead. Het opstarten van executors, het shufflen van data ertussen, het coördineren van taken. Voor kleine data is deze overhead groter dan de daadwerkelijke verwerkingstijd.

# slow for small data
df = spark.read.parquet("files/small_data.parquet")
df = df.filter(df.status == "active")
df_pandas = df.toPandas()  # converts to pandas anyway

# faster
import pandas as pd
df = pd.read_parquet("files/small_data.parquet")
df = df[df.status == "active"]

De Pandas-versie draait op één machine in het geheugen. Geen distributie-overhead. Voor datasets onder 1GB is het meestal 2-5x sneller dan Spark.

Je kunt Pandas DataFrames nog steeds opslaan als Delta tables:

# write pandas df to delta
df.to_parquet("Files/my_table", engine="pyarrow")
spark.sql("CREATE TABLE my_table USING DELTA LOCATION 'Files/my_table'")

Regel: als het past in het geheugen op één machine, gebruik Pandas. Gebruik Spark alleen wanneer je daadwerkelijk distributie nodig hebt.

Middelgrote datasets: standaardinstellingen werken prima

Voor datasets tussen 1-50GB werkt de standaard Spark configuratie in Fabric prima. Over-optimaliseer niet.

Wat op deze schaal écht belangrijk is:

1. Lees alleen de kolommen die je nodig hebt

# reads entire file
df = spark.read.parquet("files/large_table")
df = df.select("customer_id", "amount")

# reads only needed columns (way faster)
df = spark.read.parquet("files/large_table") \
    .select("customer_id", "amount")

Parquet is columnar. Het lezen van 2 kolommen uit een tabel met 50 kolommen betekent 20x minder data om te scannen.

2. Filter vroeg

# filters after loading everything
df = spark.read.parquet("files/sales")
df = df.filter(df.date >= "2024-01-01")

# filters during read (predicate pushdown)
df = spark.read.parquet("files/sales") \
    .filter("date >= '2024-01-01'")

De tweede versie leest alleen bestanden die data van 2024 bevatten. Kan hele partitions overslaan.

3. Cache alleen bij hergebruik

# loads same data twice (slow)
df = spark.table("sales")
result1 = df.filter(df.region == "US").count()
result2 = df.filter(df.region == "EU").count()

# cache when reading multiple times
df = spark.table("sales")
df.cache()
result1 = df.filter(df.region == "US").count()
result2 = df.filter(df.region == "EU").count()
df.unpersist()

Caching slaat de DataFrame op in het geheugen. Dit helpt alleen als je het daadwerkelijk hergebruikt. Cache niet als je het maar één keer gebruikt.

Voor workloads van 1-50GB zijn deze drie dingen belangrijker dan chique Spark configs.

Grote datasets: configuratie begint van belang te worden

Boven de 50GB moet je nadenken over Spark configuratie. De standaardinstellingen zijn niet geoptimaliseerd voor grootschalige implementaties.

Shuffle partitions

Dit is de belangrijkste instelling die niemand aanpast.

De standaardwaarde is 200 shuffle partitions. Dit is veel te laag voor grote data en veel te hoog voor kleine data.

# check current setting
spark.conf.get("spark.sql.shuffle.partitions")

# set based on your data size
# rule of thumb: aim for 128MB per partition

# for 10GB of data after shuffle
# 10GB / 128MB = ~80 partitions
spark.conf.set("spark.sql.shuffle.partitions", "80")

# for 100GB of data
# 100GB / 128MB = ~800 partitions  
spark.conf.set("spark.sql.shuffle.partitions", "800")

Shuffle gebeurt tijdens groupBy, join en repartition operations. Te weinig partitions betekent dat taken te groot zijn en te weinig geheugen hebben. Te veel betekent overhead door het beheren van duizenden kleine taken.

Adaptive query execution

Zet dit aan. Het staat standaard uit in sommige Fabric versies, maar maakt alles sneller.

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

AQE past het queryplan runtime aan op basis van de werkelijke datagroottes. Het combineert automatisch kleine partitions, past join strategieën aan, en beheert skew.

Zou eerlijk gezegd standaard moeten zijn, maar je moet het handmatig inschakelen.

Broadcast joins

Bij het joinen van een grote tabel met een kleine tabel kan Spark de kleine tabel broadcasten naar alle executors, in plaats van beide te shufflen.

# automatic broadcast for tables under 10MB (default)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "10485760")

# increase if you have memory
# good for dimension tables up to 100MB
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "104857600")

# or force broadcast manually
from pyspark.sql.functions import broadcast

large_df = spark.table("fact_sales")
small_df = spark.table("dim_product")

result = large_df.join(broadcast(small_df), "product_id")

Broadcasting voorkomt dure shuffles. Veel sneller voor kleine dimension tables die joinen met grote fact tables.

Dynamic partition pruning

Nog een instelling die standaard zou moeten zijn, maar niet altijd is ingeschakeld.

spark.conf.set("spark.sql.optimizer.dynamicPartitionPruning.enabled", "true")

Wanneer je filtert op een dimension table en vervolgens joint met een gepartitioneerde fact table, slaat dit het lezen van onnodige partitions over. Kan de gescande data met 10-100x verminderen.

Efficiënt lezen en schrijven

Bestandsformaten

Gebruik altijd Parquet of Delta voor analytics. Gebruik nooit CSV in productie.

# csv: slow, no schema, string types, huge files
df = spark.read.csv("files/data.csv", header=True, inferSchema=True)

# parquet: fast, schema, compressed, columnar
df = spark.read.parquet("files/data.parquet")

# delta: parquet + versioning + optimizations
df = spark.read.format("delta").load("Tables/data")

CSV is tekst. Elke keer dat je leest, worden strings geparset naar types. Parquet slaat data op in binair formaat, reeds getypeerd en gecomprimeerd.

Echte cijfers: 10GB CSV vs. 2GB Parquet met 5x snellere leestijden.

Schrijfpatronen

Hoe je data schrijft, beïnvloedt de leesprestaties voor altijd.

# creates tons of small files (slow to read later)
df.write.format("delta").mode("append").saveAsTable("my_table")

# repartition first for better file sizes
df.repartition(10).write.format("delta").mode("append").saveAsTable("my_table")

# or use coalesce if already shuffled
df.coalesce(10).write.format("delta").mode("append").saveAsTable("my_table")

Elke partition schrijft één bestand. Te veel partitions betekent duizenden kleine bestanden. Te weinig betekent gigantische bestanden die niet parallel gelezen kunnen worden.

Doel bestandsgrootte: 128MB tot 1GB per bestand, afhankelijk van de schaal.

Tabellen partitioneren

Partitioneer Delta tables op kolommen waarop je vaak filtert.

# partition by date (most common pattern)
df.write.format("delta") \
    .partitionBy("date") \
    .saveAsTable("sales")

# queries that filter on date skip entire partitions
filtered = spark.table("sales").filter("date >= '2024-01-01'")

Partitioneer alleen op kolommen met een lage cardinaliteit. Datum is goed (365 waarden). Customer ID is slecht (miljoenen waarden).

Partitioneer niet als je tabel minder dan 10GB is. De overhead is het niet waard.

Praktijkvoorbeeld: een trage notebook optimaliseren

Ik had een notebook dat 80GB aan verkoopdata verwerkte. Het duurde 45 minuten op F16 capacity. Dit is wat daadwerkelijk hielp:

Vóór optimalisatie:

# read everything
sales = spark.read.parquet("files/sales")

# filter after reading
sales = sales.filter(sales.date >= "2024-01-01")

# join to product catalog (5MB table)
products = spark.read.parquet("files/products")
result = sales.join(products, "product_id")

# aggregate
final = result.groupBy("category").agg(sum("amount"))
final.write.saveAsTable("category_sales")

Runtime: 45 minuten Capacity verbruikt: 12 CU-uren

Na optimalisatie:

# enable AQE
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.shuffle.partitions", "400")

# filter during read
sales = spark.read.parquet("files/sales") \
    .filter("date >= '2024-01-01'")

# broadcast small table
products = spark.read.parquet("files/products")
result = sales.join(broadcast(products), "product_id")

# rest same
final = result.groupBy("category").agg(sum("amount"))
final.write.saveAsTable("category_sales")

Runtime: 12 minuten Capacity verbruikt: 3.2 CU-uren

3.75x sneller door de fysica te veranderen, niet de logica. De daadwerkelijke transformaties bleven identiek.

Wat je niet kunt controleren in Fabric

In tegenstelling tot Databricks krijg je in Fabric geen volledige cluster controle. Sommige zaken worden beheerd:

Niet configureerbaar:

• Exact aantal executors
• Individuele node types
• Custom docker images
• Netwerkinstellingen
• Opslag mount points

Wel configureerbaar:

• Alle Spark configs via spark.conf.set
• Library installaties
• Omgevingsvariabelen
• Sessie timeout instellingen

Dit is eigenlijk prima voor de meeste use cases. Minder configuratie betekent minder dat kapot kan gaan. Als je atomaire controle over elke cluster instelling nodig hebt, gebruik dan Databricks in plaats daarvan.

Wanneer notebooks vs. Dataflows te gebruiken

Mensen vragen dit vaak. Hier is mijn regel:

Gebruik Dataflows Gen2 wanneer:

• Je vertrouwd bent met Power Query
• Transformaties eenvoudig zijn (filters, joins, basis aggregaties)
• Data onder de 20GB is
• Je team al Power Query kent

Gebruik notebooks wanneer:

• Je Python of complexe logica nodig hebt
• Data boven de 20GB is
• Je specifieke Spark optimalisaties nodig hebt
• Je volledige controle over de fysica wilt

Dataflows draaien ook op Spark, maar je kunt de configuratie niet aansturen. Voor grootschalig werk geven notebooks je de controle die je nodig hebt.

Opslag is ook belangrijk

Je Spark cluster is slechts de helft van de fysica. Opslag I/O is de andere helft.

Delta tables opgeslagen in OneLake gebruiken Azure Blob Storage als onderliggende laag. Dit heeft beperkingen:

• Doorvoercaps op basis van Fabric capacity
• Latentie ~10-50ms per operation
• Werkt het beste met grote sequentiële reads

Wat dit betekent:

Goed patroon:

# large scan, sequential read
df = spark.table("large_table").filter("date >= '2024-01-01'")
result = df.groupBy("category").agg(sum("amount"))

Slecht patroon:

# lots of small random reads
for customer in customer_list:
    df = spark.table("sales").filter(f"customer_id = {customer}")
    # process each customer separately

Het tweede patroon genereert duizenden kleine I/O requests. Opslag wordt de bottleneck, niet compute.

Als je data per-key moet verwerken, gebruik dan groupBy, geen loops. Laat Spark de distributie afhandelen.

Monitoren wat er daadwerkelijk gebeurt

Fabric geeft je Spark UI toegang. Gebruik het.

In je notebook na het uitvoeren van cellen:

1. Klik op de Spark application link bovenaan
2. Opent Spark UI
3. Controleer stages, tasks, data shuffle

Zoek naar:

• Tasks die veel langer duren dan andere (skew)
• Grote shuffle read/write (te veel dataverplaatsing)
• Veel tasks (te veel partitions)
• Weinig tasks (te weinig partitions)

De UI toont je de fysica. Je code vertelt je niet waarom het traag is, de uitvoeringsstatistieken wel.

Checklist voor snelle winsten

Als je notebook traag is, probeer dan deze in volgorde:

1. Controleer de datagrootte, als deze onder 1GB is, gebruik Pandas
2. Lees alleen kolommen die je nodig hebt
3. Filter zo vroeg mogelijk
4. Schakel adaptive query execution in
5. Stel shuffle partitions in op basis van datagrootte (data_gb / 0.128 = partitions)
6. Broadcast kleine tabellen in joins
7. Controleer op skew in Spark UI
8. Repartition voordat je schrijft (streef naar bestanden van 128MB-1GB)

Deze lossen 80% van de prestatieproblemen op.

Kostenoptimalisatie

Prestaties en kosten zijn verbonden in Fabric. Snellere jobs kosten minder omdat ze minder CU-uren verbruiken.

Praktijkvoorbeeld van kosten op F16 capacity:

Trage notebook:

• Runtime: 45 minuten
• CU consumptie: 12 CU-uren
• Als F16 $X/uur kost, kost dit 0.75 * X

Geoptimaliseerde notebook:

• Runtime: 12 minuten
• CU consumptie: 3.2 CU-uren
• Kosten 0.2 * X

Zelfde output, 62% lagere kosten enkel door optimalisatie.

Dit is waarom het begrijpen van de fysica belangrijk is. Je betaalt voor compute tijd. Efficiënte fysica betekent minder tijd, en dus lagere kosten.

Meer leren over Spark

Als je serieus bent over Fabric notebooks, moet je de Spark fundamentals begrijpen. Je kunt niet optimaliseren wat je niet begrijpt.

Belangrijke concepten om te leren:

• Transformaties vs. actions
• Wide vs. narrow transformaties
• Shuffle operations
• Catalyst optimizer
• Tungsten execution engine

Je hoeft geen Spark expert te zijn, maar het begrijpen van het execution model helpt je betere notebooks te schrijven.

De Lakehouse architectuur is ook van belang. Hoe je je Delta tables structureert, beïnvloedt de query prestaties net zoveel als Spark configuratie.

Tot slot

Je notebook code is slechts de helft van het verhaal. De Spark cluster die het draait, is de andere helft. De meeste mensen denken alleen aan logica en vragen zich af waarom dingen traag zijn.

Het begrijpen van de fysica, welke resources je hebt op elk SKU-niveau, en hoe je Spark correct configureert, maakt alles sneller en goedkoper.

Begin met de basis: lees minder data, filter vroeg, schakel AQE in. Ga dan verder met geavanceerdere zaken zoals shuffle partitions en broadcast joins zodra je begrijpt wat er daadwerkelijk gebeurt.

Als je nieuw bent met Fabric, bekijk dan eerst mijn introductiegids voor Power BI developers. Kom hier dan op terug wanneer je notebooks schrijft die optimalisatie nodig hebben.

De scheiding tussen logica en fysica is fundamenteel. Schrijf goede code, maar configureer de execution environment ook correct. Dan presteren notebooks pas echt goed.