Chapter 6: Distributed Processing with Spark and EMR

Driver, Cluster Manager, and Executors

A Spark cluster is like a kitchen running a banquet. The driver is the head chef who reads the orders and decides what dishes to prepare. The cluster manager is the restaurant manager who allocates kitchen stations. The executors are the line cooks who chop, sauté, and plate the food.

The Spark driver runs the application's main() function and maintains the SparkContext. It converts user actions into tasks, schedules those tasks on executors, collects results, and communicates with the cluster manager. If the driver dies, the application dies. Executors are JVM worker processes that run tasks, cache RDDs and DataFrames in memory, and send heartbeats to the driver. The cluster manager — YARN on EMR on EC2, Kubernetes on EMR on EKS and EMR Serverless, or Standalone — allocates executor processes and handles node failures.

The driver-executor split is the source of one of the most common Spark gotchas. df.collect() pulls all partitions back to the driver and can OOM it if the result is large. Conversely, you cannot reference a SparkContext inside a UDF, because the UDF runs on executors that have no access to it.

RDDs, DataFrames, and Datasets

Spark exposes three layered abstractions. The RDD is a fault-tolerant, partitioned collection of records — flexible but opaque to the optimizer. The DataFrame is a distributed table with named, typed columns; because the schema is known, Spark can rearrange operations and generate efficient code. The Dataset is a typed Scala/Java extension of DataFrame; PySpark has no separate Dataset.

If RDD is hand-written assembly, DataFrame is C — high enough that a compiler can optimize aggressively, low enough to express almost any computation. Default to DataFrames. Reach for RDDs only when you need control the optimizer cannot give you.

Catalyst Optimizer and Tungsten

Two engines make DataFrame operations fast. Catalyst is Spark's query optimizer: it translates a query into a logical plan, applies rule-based optimizations (predicate pushdown, constant folding, column pruning), then explores cost-based physical alternatives. Catalyst will push a filter down into the Parquet reader so the file format itself skips disqualified row groups. Tungsten is the physical execution engine: off-heap memory, cache-friendly binary row formats, and whole-stage code generation that fuses operators into a single tight loop of bytecode.

sales.filter("region = 'EU'") \
     .join(customers, "customer_id") \
     .groupBy("country") \
     .agg({"amount": "sum"}) \
     .show()

Catalyst pushes region = 'EU' into the sales scan, prunes customer columns, decides whether customers fits a broadcast join, and reorders operations to minimize shuffle. Tungsten then generates a single fused operator: scan → filter → project → join probe → partial aggregate, with no intermediate row collections.

Figure 6.2: Catalyst optimizer phases

Jobs, Stages, Tasks, and the Shuffle

Spark uses lazy evaluation. Transformations like filter, select, and join build a plan but do not execute. Only an action — collect(), count(), save(), write() — triggers a job.

The DAG scheduler splits each job into stages separated by shuffle boundaries. Within a stage, operators pipeline. A task is the smallest unit of work — one partition on one executor core. The task scheduler prefers data locality: PROCESS_LOCAL > NODE_LOCAL > RACK_LOCAL > ANY.

Wide transformations (groupByKey, join, distinct, repartition) force shuffles because rows sharing a key must land on the same partition. The shuffle has four phases: map-side partitioning to local disk, sorted shuffle write, network fetch by reducers, and reduce-side aggregation. Shuffle is by far the most expensive operation in Spark. If a partition is lost, Spark recomputes it from lineage — the "resilient" in RDD.

DataFrame Transformations and Actions

DataFrame operations come in two flavors: transformations that build a logical plan (lazy) and actions that trigger execution (eager). Transformations include select, filter, withColumn, groupBy, agg, join, union, distinct, repartition. Actions include show, collect, count, take, write, foreach.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date, lower, when

spark = SparkSession.builder.appName("orders-silver").getOrCreate()

raw = (spark.read
       .format("json")
       .option("multiLine", "true")
       .load("s3://lake/bronze/orders/dt=2026-05-07/"))

silver = (raw
          .filter(col("status").isNotNull())
          .withColumn("order_date", to_date(col("created_at")))
          .withColumn("region", lower(col("region")))
          .withColumn("is_refund", when(col("amount") < 0, True).otherwise(False))
          .dropDuplicates(["order_id"]))

(silver.write
       .format("parquet")
       .mode("overwrite")
       .partitionBy("order_date", "region")
       .save("s3://lake/silver/orders/"))

The whole pipeline is a chain of transformations that build a single logical plan; nothing runs until .save(). Output is partitioned by date and region so downstream queries can skip files via partition pruning.

Spark SQL and the Hive Metastore

Spark SQL is the same engine as DataFrame with SQL as another front end. Any DataFrame can be registered as a temporary view (df.createOrReplaceTempView("orders")) and queried with SQL; any SQL query returns a DataFrame.

For long-lived tables Spark uses a metastore — typically the Hive Metastore or AWS Glue Data Catalog on EMR — to persist schema, location, partition layout, and storage format. EMR integrates with the AWS Glue Data Catalog, so the same table is visible to Spark, Athena, Redshift Spectrum, and Trino. That single shared catalog is what lets a pile of S3 files behave like a queryable warehouse for multiple engines.

PySpark vs Scala vs SQL

DataFrame and SQL operations run inside the JVM regardless of front-end language, so PySpark matches Scala speed for declarative work. PySpark falls behind only on custom Python UDFs, which serialize every row to Python and back. Pandas UDFs (vectorized, Arrow-based) close most of that gap.

Pragmatic recommendation: PySpark for orchestration and team familiarity, Spark SQL views for reusable transformations, and Scala libraries for performance-critical components.

Amazon EMR — Three Deployment Models

Amazon EMR is AWS's managed big-data platform. It supports Spark, Hadoop, Hive, HBase, Trino, and Flink, but for modern data engineering it is shorthand for "managed Spark." AWS reports EMR Spark runs up to 5.4x faster than open-source Apache Spark.

EMR on EC2 (Classic Clusters)

A cluster of EC2 instances configured as Hadoop master and worker nodes, with YARN as the cluster manager and HDFS or EMRFS (S3) for storage. Maximum configurability, manual cluster sizing, 3–10 minute cold start. Spot Instance integration enables 50–90% cost reductions for non-critical workloads.

Choose when: long-running 24/7 ETL, configuration-heavy workloads (custom JARs, kernel tuning, mixed frameworks), or sustained batch where Spot Instances on a persistent cluster pay for themselves.

EMR Serverless

You define an "application" with a framework (Spark or Hive) and submit jobs; AWS provisions and tears down capacity transparently. Pay-per-job billing based on actual vCPU-hours and memory-hours — no idle cost. Cold start of 30–60 seconds, fully automatic autoscaling, limited customization (custom Python venvs and Docker images supported).

Choose when: bursty or scheduled workloads, dev/test environments, anywhere a long-lived cluster would waste 23+ hours of idle capacity per day.

EMR on EKS

Spark jobs run as Kubernetes pods on an existing Amazon EKS cluster. Native K8s scheduling, RBAC, network policies. You pay for the EKS cluster regardless of EMR usage; EMR adds a per-vCPU surcharge for Spark jobs. Resource sharing across workload types improves utilization.

Choose when: the organization already runs EKS in production and wants a unified platform for data and application workloads, or needs advanced K8s features like GPU node pools.

Comparison

Mature platforms often mix the models: EMR on EC2 (or EKS) for always-on baseline pipelines, EMR Serverless for the long tail of analyst queries, backfills, and data-science exploration. This minimizes idle cost while preserving operational control where it matters.

Partition Sizing and Skew

Partitions are the unit of parallelism. Sizing rule of thumb: target 100–200 MB per partition. Tasks then run for tens of seconds — long enough to amortize scheduling overhead, short enough that retries are cheap. The default spark.sql.shuffle.partitions = 200 is rarely the right answer for real workloads.

Skew is the silent killer. If most join keys have a few thousand rows but one (say customer_id = 'enterprise_account') has 50 million, that one partition becomes a "straggler" while the rest of the cluster idles. The task duration histogram in the Spark UI surfaces this immediately.

Manual mitigations include repartitioning before joins, salting hot keys (append a random suffix and replicate the matching side), filtering early, and pre-aggregating before the shuffle. AQE's OptimizeSkewedJoin rule handles this automatically — published benchmarks show reductions from ~7.7 minutes to ~1 minute on heavily skewed joins.

Broadcast Joins and Shuffle Reduction

A regular join shuffles both sides so matching keys land on the same partition — expensive. A broadcast join sends the smaller table to every executor as a hash table; each executor probes its local table with no shuffle on the large side.

Figure 6.4: Shuffle hash join vs broadcast join

Broadcast joins win when one side is small (default <30 MB after compression, configurable via spark.sql.adaptive.broadcastJoinThreshold) or when the large side is heavily skewed — broadcast joins are immune to skew because they do not partition the large side at all.

from pyspark.sql.functions import broadcast

# Explicit hint - recommended when you know the table is small
sales.join(broadcast(dim_region), "region_id")

# Automatic via AQE - converts at runtime when size < threshold
# Requires: spark.sql.adaptive.enabled = true

The risk: broadcast joins OOM the driver if the "small" side turns out to be 5 GB. Validate the size before using an explicit hint, and rely on AQE's runtime size detection when unsure.

Caching, Persisting, and Adaptive Query Execution

Caching keeps a DataFrame's partitions in memory (or memory + disk) so subsequent actions reuse them. Cache when a DataFrame is referenced by multiple actions, when an iterative algorithm revisits the same data, or when an expensive transformation feeds multiple queries. Don't cache one-time DataFrames or data that doesn't fit. Always pair .cache() with .unpersist().

Adaptive Query Execution (AQE), default-enabled in modern Spark, is the most important tuning feature added in the last decade. AQE uses runtime statistics — actual intermediate result sizes after each stage — to re-plan the rest of the query. The shift is from static optimization (plans set before execution) to dynamic optimization (plans adapt to runtime data sizes).

Figure 6.5: Adaptive Query Execution — runtime re-planning loop

A solid baseline configuration for production Spark on EMR:

# Enable AQE and its sub-rules
spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.adaptive.skewJoin.enabled=true
spark.sql.adaptive.localShuffleReader.enabled=true

# Broadcast join threshold (after AQE runtime sizing)
spark.sql.adaptive.broadcastJoinThreshold=30M

# Dynamic partition pruning (for partitioned fact tables)
spark.sql.dynamicPartitionPruning.enabled=true

# Initial shuffle partition count (AQE will coalesce as needed)
spark.sql.shuffle.partitions=400

Validation Workflow

1. Enable AQE and run the job; capture runtime in the Spark UI.
2. Check the SQL tab for AQE annotations showing coalesced partitions or converted joins.
3. Look for straggler tasks in the Stages tab; add broadcast() hints or salt if AQE missed a skew.
4. Check executor memory and GC time; spilling to disk means more memory or smaller partitions.
5. Use EXPLAIN FORMATTED to confirm predicate pushdown to file readers.