Spark: Basics and Performance Tuning

Apache Spark has become the de facto standard for large-scale data processing, powering analytics at Netflix, Uber, Airbnb, and thousands of enterprises processing petabytes of data daily. Unlike Hadoop MapReduce (which writes intermediate results to disk between stages), Spark's in-memory computing engine delivers 10-100× faster processing for iterative algorithms, interactive queries, and streaming workloads — while providing a unified API for batch processing, real-time streaming, machine learning, and graph analytics.

However, Spark's performance advantage is not automatic — poorly configured jobs can be slower and more expensive than MapReduce equivalents. The difference between a 10-minute job and a 2-hour job often comes down to partitioning strategy, shuffle management, memory configuration, and data format selection. This guide covers Spark's architecture, data abstraction layers (RDDs, DataFrames, Datasets), and the specific tuning techniques that transform underperforming jobs into optimised pipelines — from partition sizing and broadcast joins through memory management, GC tuning, and Spark UI diagnostics.

Understand Spark's execution model to identify optimisation opportunities:

• Driver Program: The master process that creates the SparkContext, defines transformations and actions, builds the Directed Acyclic Graph (DAG), and coordinates task scheduling across the cluster. The driver maintains metadata about partitions, cached data, and accumulator values. Driver memory (spark.driver.memory) typically needs 2-8GB — increase for jobs collecting large results to the driver.
• Cluster Manager: YARN (most common in Hadoop ecosystems), Kubernetes (growing for cloud-native deployments), Mesos, or Standalone mode. YARN manages resource allocation — request executors with specific memory and core counts. Kubernetes provides containerised Spark with auto-scaling, namespace isolation, and better resource utilisation for multi-tenant clusters.
• Executors: Worker processes that run tasks and cache data. Each executor has dedicated memory (spark.executor.memory) and CPU cores (spark.executor.cores). Configure executor sizing based on workload: memory-intensive jobs (large joins, caching) need more memory per executor; CPU-intensive jobs (complex transformations, ML training) benefit from more cores.
• DAG Scheduler: Spark decomposes jobs into stages separated by shuffle boundaries. Within each stage, tasks execute in parallel across partitions. The DAG scheduler optimises by pipelining narrow transformations (map, filter, flatMap) into single tasks — only wide transformations (groupBy, join, repartition) create stage boundaries requiring data exchange. Minimising stage boundaries is key to performance.
• Task Execution: Each task processes one partition of data. Task parallelism equals the number of partitions — too few partitions under-utilise the cluster, too many create scheduling overhead. Target 2-4 tasks per available core with partition sizes of 128-256MB for optimal throughput. The Spark UI's "Stages" tab reveals task distribution and identifies stragglers.

Select the appropriate data abstraction for your workload characteristics:

• RDDs (Resilient Distributed Datasets): The foundational abstraction — untyped, distributed collections with functional transformations (map, filter, reduce). RDDs provide full control over data layout and operations but bypass Spark's Catalyst optimiser and Tungsten execution engine. Use RDDs only for unstructured data processing, custom partitioning logic, or low-level operations that DataFrames cannot express.
• DataFrames: Structured data with named columns and schema — equivalent to database tables. DataFrames leverage the Catalyst Optimiser (predicate pushdown, join reordering, constant folding) and Tungsten Engine (off-heap memory management, code generation, cache-aware computation). DataFrames are typically 2-10× faster than equivalent RDD operations due to these optimisations. Use DataFrames for structured/semi-structured data processing, SQL queries, and ETL pipelines.
• Datasets: Typed DataFrames — combine compile-time type safety (Scala/Java) with Catalyst/Tungsten optimisations. Datasets use encoders for efficient serialisation (binary format, not Java serialisation). The performance difference between DataFrames and Datasets is minimal in most cases. Use Datasets when type safety is important (complex domain objects, library APIs).
• Migration Path: Convert legacy RDD code to DataFrames for automatic performance improvements — rdd.toDF() or spark.createDataFrame(rdd, schema). Most RDD operations have DataFrame equivalents: rdd.map() → df.select()/df.withColumn(), rdd.filter() → df.where(), rdd.groupBy() → df.groupBy(). The Catalyst optimiser handles operation reordering that RDD users must implement manually.
• PySpark Considerations: PySpark RDD operations serialize data between JVM and Python processes — causing 2-10× overhead compared to Scala/Java. PySpark DataFrames avoid this overhead because operations execute in the JVM via Catalyst. Always prefer DataFrames in PySpark — use pandas_udf (Vectorized UDFs) instead of regular Python UDFs for custom operations.

Optimise data distribution across the cluster to maximise parallelism and minimise shuffle:

• Partition Sizing: Target 128-256MB per partition for optimal I/O and memory utilisation. Calculate: numPartitions = dataSize / 128MB. Too few partitions (e.g., 10 partitions for 100GB) create large tasks that risk OOM errors and under-utilise cluster cores. Too many partitions (e.g., 100,000 partitions for 1GB) create scheduling overhead and many small files in output.
• Repartition vs Coalesce: Use repartition(n) to increase partitions or redistribute data evenly (triggers a full shuffle). Use coalesce(n) to decrease partitions without shuffle — it merges adjacent partitions, but can create uneven partition sizes. For reducing partitions after a filter that eliminates most data, coalesce() is significantly faster than repartition().
• Hash Partitioning: repartition(n, col("key")) distributes data by hash of specified columns — ensures all records with the same key land in the same partition. Critical for join optimisation: if both sides of a join are hash-partitioned on the join key, Spark performs a partition-local join without shuffle (sort-merge join with co-located partitions).
• Range Partitioning: repartitionByRange(n, col("key")) distributes data by value ranges — partition 1 gets keys A-D, partition 2 gets E-H, etc. Useful for ordered output, range-based queries, and avoiding skew when hash partitioning creates uneven distribution (e.g., one key value dominates).
• Adaptive Query Execution (AQE): Spark 3.x's AQE automatically adjusts partition counts and join strategies at runtime based on actual data statistics. Enable with spark.sql.adaptive.enabled=true. AQE coalesces small post-shuffle partitions (spark.sql.adaptive.coalescePartitions.enabled), switches to broadcast join when one side is small (spark.sql.adaptive.localShuffleReader.enabled), and handles skew joins automatically.

Shuffle operations are the most expensive operation in Spark — minimise them relentlessly:

• Understanding Shuffle: Shuffle occurs when data must be redistributed across partitions — triggered by groupBy(), join(), repartition(), distinct(), and orderBy(). Shuffle writes intermediate data to disk, transfers it over the network, and reads it back — involving serialisation, disk I/O, network I/O, and deserialisation. A single shuffle stage can dominate total job execution time.
• Broadcast Joins: When one side of a join is small enough to fit in executor memory (default threshold: 10MB, configurable via spark.sql.autoBroadcastJoinThreshold), Spark broadcasts the small table to all executors — eliminating shuffle entirely. Increase the threshold to 100-500MB for medium tables: spark.sql.autoBroadcastJoinThreshold=256m. Force broadcast with broadcast(df) hint.
• Map-Side Aggregation: Use reduceByKey() instead of groupByKey() for RDD aggregations — reduceByKey() performs partial aggregation on each partition before shuffle, dramatically reducing data transfer. For DataFrames, Spark's Catalyst optimiser automatically applies partial aggregation for groupBy().agg() operations.
• Shuffle Partition Count: The default spark.sql.shuffle.partitions=200 is rarely optimal. Too few partitions create large shuffle blocks and potential OOM; too many create excessive small files. Calculate: shufflePartitions = shuffleDataSize / 128MB. With AQE enabled, Spark auto-coalesces small partitions, making the initial count less critical.
• Skew Handling: Data skew (one partition 100× larger than others) causes one task to run for hours while others finish in seconds. Solutions: salting (add random prefix to skewed keys, join, then aggregate), AQE skew join handling (spark.sql.adaptive.skewJoin.enabled=true), or custom repartitioning to split hot keys across multiple partitions.

Configure Spark's unified memory model to prevent OOM errors and minimise GC overhead:

• Unified Memory Model: Spark divides executor memory into execution memory (shuffles, joins, sorts, aggregations) and storage memory (cached RDDs/DataFrames). The split is dynamic — execution can borrow from storage and vice versa via spark.memory.fraction (default 0.6 = 60% of heap for Spark operations, 40% for user data structures and internal metadata). Increase to 0.7-0.8 for jobs with heavy caching or complex aggregations.
• Off-Heap Memory: Enable off-heap memory (spark.memory.offHeap.enabled=true, spark.memory.offHeap.size=4g) to move storage and execution data outside the JVM heap — reducing GC pressure significantly. Off-heap memory is managed by Tungsten's memory manager with explicit allocation/deallocation. Particularly effective for jobs with large caches or many concurrent tasks competing for heap space.
• Garbage Collection Tuning: For Spark workloads, use G1GC (-XX:+UseG1GC) with appropriate region size: -XX:G1HeapRegionSize=16m for large heaps (>16GB). Monitor GC time in Spark UI — if GC exceeds 10% of task time, increase executor memory or reduce the number of objects created. Consider ZGC (-XX:+UseZGC) for Java 17+ with large heaps (>32GB) for sub-millisecond pause times.
• Executor Sizing: Avoid monolithic executors (single executor with all cluster memory) — large heaps cause long GC pauses. Target 4-8 cores and 16-32GB memory per executor. For YARN: spark.executor.instances × spark.executor.memory ≤ cluster memory with overhead (spark.executor.memoryOverhead, default 10% of executor memory, minimum 384MB).
• Dynamic Allocation: Enable dynamic allocation (spark.dynamicAllocation.enabled=true) to automatically add/remove executors based on workload — prevents resource waste during idle periods and scales up for demanding stages. Configure spark.dynamicAllocation.minExecutors, maxExecutors, and executorIdleTimeout to balance responsiveness with resource efficiency.

Maximise Spark SQL performance through data formats, query patterns, and optimiser hints:

• Catalyst Optimiser: Spark SQL's query optimiser applies rule-based and cost-based optimisations — predicate pushdown (filter before join), projection pruning (read only needed columns), join reordering (smallest tables first), constant folding, and null propagation. Write queries using DataFrame API or SQL — both pass through Catalyst. Avoid UDFs when possible — they're opaque to the optimiser and prevent predicate pushdown.
• Data Format Selection: Parquet (columnar, compressed, with predicate pushdown and column pruning) is the default recommendation — 2-10× faster reads than CSV/JSON for analytical queries. ORC performs similarly with better ACID support in Hive environments. Delta Lake adds ACID transactions, time travel, and schema evolution on top of Parquet. Avoid row-oriented formats (CSV, JSON, Avro) for analytical workloads.
• Predicate Pushdown: Write filters before joins and aggregations — df.where("date > '2024-01-01'").join(other) instead of df.join(other).where("date > '2024-01-01'"). With Parquet/ORC, predicate pushdown skips entire row groups/stripes that don't match filter conditions — reading 1% of data instead of 100% for selective queries.
• Bucketing: Pre-partition data by join keys during write: df.write.bucketBy(256, "user_id").saveAsTable("users"). Subsequent joins on user_id avoid shuffle entirely — Spark recognises that data is already co-partitioned. Bucketing is most effective for repeatedly joined large tables in data warehouse patterns.
• Cost-Based Optimiser (CBO): Enable CBO (spark.sql.cbo.enabled=true) and compute table statistics (ANALYZE TABLE table COMPUTE STATISTICS FOR ALL COLUMNS) for intelligent join strategy selection. CBO uses row counts, column cardinality, null counts, and min/max values to choose between broadcast, sort-merge, and shuffle-hash joins — and to determine optimal join order for multi-table queries.

Implement observability and operational practices for reliable Spark pipelines:

• Spark UI Analysis: The Spark UI (port 4040) provides DAG visualisation, stage/task metrics, and executor diagnostics. Key metrics to monitor: task duration distribution (identify stragglers), shuffle read/write sizes (identify expensive shuffles), GC time per task (identify memory pressure), and input/output sizes per stage (identify data skew). Use the SQL tab for query plan inspection.
• Spark History Server: Configure Spark History Server for post-mortem analysis of completed jobs — essential for debugging production failures. Store event logs in HDFS or S3 (spark.eventLog.enabled=true, spark.eventLog.dir=s3://spark-logs/). Retain logs for 30-90 days for trend analysis and performance regression detection.
• Metrics Integration: Export Spark metrics to Prometheus/Grafana or Datadog for real-time dashboarding — executor memory usage, task rates, shuffle volumes, and streaming batch durations. Set alerts for job duration regression (>2× historical average), high GC ratios (>15% of task time), and executor loss events.
• Data Quality Checks: Integrate data quality validation into pipelines — row count assertions, schema validation, null percentage checks, and statistical distribution monitoring. Use Great Expectations or custom validation frameworks. Fail pipelines early on data quality violations rather than producing incorrect results downstream.
• Production Configuration: Essential production settings: spark.sql.adaptive.enabled=true (AQE), spark.serializer=org.apache.spark.serializer.KryoSerializer (10× faster than Java serialisation), spark.sql.parquet.filterPushdown=true, spark.speculation=true (re-launch slow tasks on other executors), and appropriate retry settings (spark.task.maxFailures=4). Use Delta Lake or Iceberg for ACID guarantees in production data lakes.