Apache Spark Essentials

Spark is a distributed processing engine for large datasets that don't fit (or don't process fast enough) on a single machine. It splits work across a cluster, keeps intermediate data in memory when possible, and gives you the same programming model whether you're processing 1GB locally or 10TB on a hundred nodes. It's the default choice for batch ETL, large-scale feature engineering, and increasingly for streaming pipelines via Structured Streaming.

RDDs vs DataFrames

RDDs (Resilient Distributed Datasets) are Spark's original abstraction: an immutable, partitioned collection of objects spread across the cluster, rebuilt from its lineage graph if a node fails. They give you full control but no query optimization — Spark just runs whatever operations you write, in order. DataFrames sit on top of RDDs and add a schema plus a query optimizer (Catalyst). Almost all new code should use DataFrames (or the typed Dataset API in Scala/Java); reach for RDDs only for unstructured data or custom partitioning logic the DataFrame API can't express.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, IntegerType

spark = (
    SparkSession.builder
    .appName("orders-etl")
    .config("spark.sql.shuffle.partitions", "200")
    .getOrCreate()
)

# RDD: low-level, manual transformations
rdd = spark.sparkContext.textFile("s3://bucket/orders/*.csv")
header = rdd.first()
parsed = (
    rdd.filter(lambda line: line != header)
    .map(lambda line: line.split(","))
    .map(lambda cols: (cols[0], float(cols[2])))
)

# DataFrame: schema-aware, optimized by Catalyst
schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("customer_id", StringType(), False),
    StructField("amount", DoubleType(), False),
    StructField("quantity", IntegerType(), False),
])

orders = (
    spark.read
    .schema(schema)
    .option("header", True)
    .csv("s3://bucket/orders/*.csv")
)

orders.printSchema()
orders.show(5, truncate=False)

Transformations vs Actions (Lazy Evaluation)

Every Spark operation is either a transformation or an action. Transformations (select, filter, groupBy, join, withColumn) are lazy — they build up a logical plan but don't touch data. Actions (show, collect, count, write) trigger actual execution: Spark compiles the accumulated transformations into a physical plan and runs it across the cluster. Understanding this is critical for debugging performance — a slow line of code you 'notice' is often just where the accumulated laziness finally executes, not necessarily the expensive step itself.

# Chain of transformations — nothing executes yet
high_value = (
    orders
    .filter(F.col("amount") > 100)
    .withColumn("amount_usd", F.round(F.col("amount"), 2))
    .groupBy("customer_id")
    .agg(
        F.sum("amount_usd").alias("total_spent"),
        F.count("order_id").alias("order_count"),
    )
    .orderBy(F.desc("total_spent"))
)

# .explain() shows the physical plan Spark WOULD run — still no execution
high_value.explain(mode="formatted")

# Actions trigger the actual job
top_20 = high_value.limit(20).collect()  # pulls results to the driver — use sparingly
high_value.count()                        # triggers a full pass
high_value.write.mode("overwrite").parquet("s3://bucket/reports/top_customers/")

# Cache when you'll reuse the same DataFrame across multiple actions
high_value.cache()
high_value.count()          # first action materializes the cache
high_value.filter(F.col("order_count") > 5).show()  # reuses cached data, no recompute

Partitioning & Shuffles

Data is split into partitions, and each partition is processed by one task on one executor core. Too few partitions and you underuse the cluster; too many and scheduling overhead dominates. Operations like groupBy, join, and distinct require a shuffle — data gets redistributed across the network so matching keys land on the same partition. Shuffles are the single most expensive thing in Spark: they hit disk, network, and serialization all at once. Minimizing and right-sizing shuffles is most of what 'Spark performance tuning' actually means.

# Repartition when writing out — controls output file count and shuffle parallelism
orders.repartition(50, "customer_id").write.parquet("s3://bucket/orders_by_customer/")

# coalesce avoids a full shuffle when you're only reducing partition count
small_result.coalesce(1).write.csv("s3://bucket/report.csv")

# Broadcast join — avoids shuffling the large table when one side is small
from pyspark.sql.functions import broadcast

regions = spark.read.parquet("s3://bucket/regions/")  # small lookup table
enriched = orders.join(broadcast(regions), "region_id", "left")

# spark.sql.autoBroadcastJoinThreshold controls the size cutoff for auto-broadcast
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 10 * 1024 * 1024)  # 10MB

# Salting a skewed join key to spread a hot key across more partitions
from pyspark.sql.functions import rand, concat, lit, floor

salted = orders.withColumn(
    "salted_key", concat(F.col("customer_id"), lit("_"), floor(rand() * 10))
)

Spark SQL

DataFrames and SQL are two syntaxes for the same Catalyst-optimized plan — you can freely mix them, and there's no performance penalty for preferring SQL. Registering a DataFrame as a temp view lets analysts and downstream tools query it with plain SQL.

-- Registered via: orders.createOrReplaceTempView("orders")
SELECT
  customer_id,
  DATE_TRUNC('month', order_date) AS month,
  SUM(amount) AS total_spent,
  COUNT(*) AS order_count,
  RANK() OVER (PARTITION BY DATE_TRUNC('month', order_date) ORDER BY SUM(amount) DESC) AS rank
FROM orders
WHERE amount > 0
GROUP BY customer_id, DATE_TRUNC('month', order_date)
HAVING COUNT(*) >= 3
ORDER BY month, rank
LIMIT 100

Cluster Architecture

A Spark application has one driver process, which runs your main program, builds the execution plan, and schedules tasks — and multiple executor processes, which run on worker nodes, execute tasks in parallel, and cache data. A cluster manager (YARN, Kubernetes, or Spark's own standalone mode) allocates the executors. Managed platforms (Databricks, EMR, Dataproc) handle this provisioning for you; understanding it still matters for sizing (executor memory/cores) and debugging OOM errors, which almost always trace back to a driver collecting too much data or an executor holding too large a partition.

# spark-submit — typical cluster sizing flags
# spark-submit \
#   --master yarn \
#   --deploy-mode cluster \
#   --num-executors 20 \
#   --executor-cores 4 \
#   --executor-memory 8g \
#   --driver-memory 4g \
#   --conf spark.sql.shuffle.partitions=400 \
#   etl_job.py

# Adaptive Query Execution (AQE) — lets Spark re-plan at runtime using
# actual data statistics instead of only the pre-execution estimate
spark.conf.set("spark.sql.adaptive.enabled", True)
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", True)
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", True)

Gotchas & Tips

• Never call .collect() on a large DataFrame — it pulls every row to the driver's memory and is the most common cause of driver OOM crashes. Use .show(), .take(n), or write to storage instead.

• Avoid Python UDFs when a built-in Spark SQL function will do — UDFs serialize data row-by-row between the JVM and the Python process, which is dramatically slower than the vectorized, in-JVM execution of native functions.

• Data skew (one key vastly outnumbers the others) causes one task to run far longer than the rest — you'll see a job 'stuck' at 199/200 tasks. Salting the key or enabling AQE's skew join handling usually fixes it.

• Unpersist cached DataFrames (.unpersist()) once you're done with them — cached data counts against executor memory and can push out data you actually still need.

• Prefer Parquet (columnar, compressed, splittable) over CSV/JSON for anything read more than once — it's both smaller on disk and lets Spark skip columns and row groups it doesn't need.

• spark.sql.shuffle.partitions defaults to 200 regardless of data size — tune it down for small datasets (fewer, larger partitions) and up for very large ones, or just enable AQE and let Spark adjust it automatically.