Apache Spark Cheatsheet

The DataFrame API, the join strategies, and the partitioning/caching knobs that actually move the needle on a real cluster. For the full explanations, see the Spark guides.

Architecture in one picture

Every Spark job maps onto the same three pieces, and most “why is this slow” questions trace back to one of them.

The driver never touches your data directly — it plans stages and ships tasks. Executors do the actual reading, transforming, and shuffling. When a job hangs, check the Spark UI’s Stages tab first: a single long-running task inside a stage almost always means skew, not a slow cluster.

Starting a session

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("orders-pipeline")
    .config("spark.sql.shuffle.partitions", "200")
    .config("spark.sql.adaptive.enabled", "true")
    .getOrCreate()
)

spark.sql.shuffle.partitions defaults to 200 regardless of cluster size or data volume — that default is tuned for nothing in particular. With AQE (Adaptive Query Execution) enabled in Spark 3.x, the engine coalesces shuffle partitions automatically after seeing real data sizes, which is why spark.sql.adaptive.enabled=true is worth turning on before you touch the partition count manually.

RDD vs DataFrame vs Dataset

In practice: write DataFrame code by default. RDDs cost you the Catalyst optimizer and Tungsten’s binary row format, which is where most of Spark’s actual performance comes from — dropping to RDDs should be a deliberate exception (custom partitioners, low-level control), not a starting point.

Choosing a file format

Landing raw data as CSV/JSON and converting to Parquet (or Delta) as the first pipeline step is a standard pattern — it moves the “read the whole file just to get two columns” tax to a one-time conversion instead of paying it on every downstream query.

Reading and writing data

df = spark.read.parquet("s3://bucket/orders/")
df = spark.read.option("header", True).option("inferSchema", True).csv("s3://bucket/raw/orders.csv")
df = spark.read.option("multiline", True).json("s3://bucket/events/")

# Explicit schema — skips a full data scan just to infer types
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType
schema = StructType([
    StructField("order_id", StringType()),
    StructField("amount", DoubleType()),
    StructField("created_at", TimestampType()),
])
df = spark.read.schema(schema).parquet("s3://bucket/orders/")

(df.write
   .mode("overwrite")
   .partitionBy("order_date")
   .parquet("s3://bucket/orders_clean/"))

inferSchema triggers a second full pass over the data just to guess types — fine for a 10 MB CSV in a notebook, a real cost on a multi-GB file. Parquet stores its own schema, so inferSchema is a non-issue there; it’s a CSV/JSON-specific tax.

Core DataFrame transformations

from pyspark.sql import functions as F

(df
 .filter(F.col("status") == "completed")
 .withColumn("amount_usd", F.col("amount") * F.col("fx_rate"))
 .withColumn("order_month", F.date_trunc("month", F.col("created_at")))
 .select("order_id", "customer_id", "amount_usd", "order_month")
 .groupBy("order_month")
 .agg(
     F.sum("amount_usd").alias("revenue"),
     F.countDistinct("customer_id").alias("unique_customers"),
 )
 .orderBy("order_month"))

Transformations (filter, withColumn, select, groupBy) are lazy — nothing runs until an action (.show(), .collect(), .write(), .count()) triggers the DAG. This is why a chain of ten .withColumn() calls costs nothing by itself; the cost shows up entirely at the action, which is also why .explain() on the DataFrame before the action is the right place to look for problems.

Column expressions worth knowing

df.withColumn("is_high_value", F.when(F.col("amount") > 500, True).otherwise(False))
df.withColumn("clean_email", F.lower(F.trim(F.col("email"))))
df.withColumn("tags_array", F.split(F.col("tags"), ","))
df.withColumn("first_tag", F.col("tags_array")[0])
df.na.fill({"amount": 0, "region": "unknown"})
df.na.drop(subset=["customer_id"])

Joins — the decision that matters most

# Inner, left, right, full, left_semi, left_anti
df_orders.join(df_customers, on="customer_id", how="left")
df_orders.join(df_returns, on="order_id", how="left_anti")   # orders with NO matching return

Broadcast join — the single biggest join optimization

from pyspark.sql.functions import broadcast

df_orders.join(broadcast(df_small_dim_table), on="product_id")

When one side of a join is small enough to fit in executor memory (Spark’s default threshold is 10 MB, controlled by spark.sql.autoBroadcastJoinThreshold), broadcasting it to every executor avoids a full shuffle of the large table entirely. This turns a shuffle-heavy sort-merge join into a shuffle-free map-side join. Spark tries to do this automatically under the threshold, but explicit broadcast() hints matter once your dimension table is a CTE or the result of a filter, since Spark’s size estimate can be wrong for derived DataFrames.

Handling skewed joins

If one join key (say, a single customer_id = NULL bucket, or one mega-retailer in a store_id join) dominates the data, one task ends up doing 90% of the work while every other task finishes in seconds. Two practical fixes:

# 1. Salt the skewed key to spread it across more partitions
df_orders_salted = df_orders.withColumn("salt", (F.rand() * 10).cast("int"))
df_dim_salted = df_dim.withColumn("salt", F.explode(F.array([F.lit(i) for i in range(10)])))
df_orders_salted.join(df_dim_salted, ["store_id", "salt"])

# 2. Let AQE handle it (Spark 3.x, on by default with spark.sql.adaptive.skewJoin.enabled)
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

AQE’s skew join handling should be your first move — it splits oversized partitions automatically at runtime based on actual post-shuffle statistics. Manual salting is the fallback for Spark 2.x clusters or cases AQE doesn’t catch.

Partitioning: repartition vs coalesce

df.repartition(200)                       # full shuffle, can increase or decrease partitions
df.repartition(200, "customer_id")        # shuffle, partitioned by key — good before a groupBy/join
df.coalesce(50)                           # no shuffle, can only decrease partitions

coalesce avoids a shuffle by merging existing partitions in place, which is why it can’t increase partition count and why it can produce uneven partition sizes if the input was already skewed. repartition always shuffles but gives you an even, deterministic distribution — use it after a heavy filter has left you with mostly-empty partitions, and use coalesce right before a write when you just want fewer output files without paying for a shuffle.

Caching and persistence

df.cache()                                  # shorthand for persist(MEMORY_AND_DISK)
df.persist(StorageLevel.MEMORY_ONLY)
df.unpersist()

Caching only pays off when the same DataFrame is used by more than one downstream action — caching something you touch exactly once just adds overhead. A common miss: caching before a filter that would have shrunk the data first. Filter, then cache, then branch into multiple aggregations.

Spark SQL

df.createOrReplaceTempView("orders")
spark.sql("""
    SELECT customer_id, DATE_TRUNC('month', created_at) AS month, SUM(amount) AS revenue
    FROM orders
    WHERE status = 'completed'
    GROUP BY 1, 2
""").show()

DataFrame and SQL compile to the identical Catalyst logical plan — there’s no performance difference between the two, so pick whichever reads more clearly for the transformation in front of you. Mixing both in the same pipeline (SQL for a gnarly join, DataFrame API for the rest) is normal and costs nothing.

Window functions in Spark

from pyspark.sql.window import Window

w = Window.partitionBy("customer_id").orderBy(F.col("order_date").desc())
df.withColumn("rn", F.row_number().over(w)).filter(F.col("rn") == 1)   # latest order per customer

w_running = Window.partitionBy("customer_id").orderBy("order_date").rowsBetween(Window.unboundedPreceding, Window.currentRow)
df.withColumn("running_total", F.sum("amount").over(w_running))

Same semantics as SQL window functions — Window.unboundedPreceding / Window.currentRow map directly onto ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW. See the SQL window functions cheatsheet for the underlying frame logic.

UDFs — last resort, not first choice

from pyspark.sql.types import IntegerType

@F.udf(returnType=IntegerType())
def days_since(ts):
    from datetime import datetime
    return (datetime.now() - ts).days

df.withColumn("age_days", days_since(F.col("created_at")))

Python UDFs serialize every row out of the JVM into a Python process and back, which bypasses Tungsten’s optimized in-JVM execution entirely — they’re routinely 10-100x slower than an equivalent built-in function. Before writing a UDF, check pyspark.sql.functions for a native equivalent (F.datediff(F.current_date(), F.col("created_at")) replaces the example above with zero serialization cost). If a UDF is unavoidable, a Pandas UDF (@F.pandas_udf) at least vectorizes the Python side via Arrow instead of row-by-row.

Reading an explain plan

df.explain(mode="formatted")

Look for Exchange nodes — each one is a shuffle. A plan with three stacked Exchange nodes for what looks like a simple join usually means an unnecessary repartition upstream, or a join key that doesn’t match existing partitioning. BroadcastHashJoin in the plan confirms a join was broadcast; SortMergeJoin confirms it wasn’t, and is your cue to check whether the smaller side should have been.

Executor sizing quick reference

# spark-submit / cluster config
--num-executors 20
--executor-cores 4
--executor-memory 16g
--conf spark.executor.memoryOverhead=2g

Too many cores per executor with too little memory per core is the most common misconfiguration — it looks like “the cluster is small” when it’s actually “each task has too little memory to spill cleanly and is thrashing disk instead.”

Common errors and what they actually mean

Common patterns

Deduplicate, keeping the latest record

w = Window.partitionBy("order_id").orderBy(F.col("updated_at").desc())
df_dedup = (df
    .withColumn("rn", F.row_number().over(w))
    .filter(F.col("rn") == 1)
    .drop("rn"))

Slowly changing dimension (SCD Type 1) merge with Delta Lake

from delta.tables import DeltaTable

target = DeltaTable.forPath(spark, "s3://bucket/dim_customer/")
(target.alias("t")
    .merge(df_updates.alias("s"), "t.customer_id = s.customer_id")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute())

Reading incrementally with a watermark

df_stream = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "orders")
    .load())

(df_stream
    .withWatermark("event_time", "10 minutes")
    .groupBy(F.window("event_time", "5 minutes"), "region")
    .agg(F.sum("amount"))
    .writeStream.outputMode("update").format("console").start())

The watermark tells Spark how long to wait for late-arriving data before finalizing a window — set it too tight and you drop legitimately late events; too loose and state grows unbounded.

Not covered: MLlib, GraphX, and low-level RDD partitioner APIs aren’t included here — this sheet is scoped to the DataFrame/SQL engine, which is what most day-to-day Spark work touches. For deeper dives on any section above, see the Spark guides.