Don't spend months learning Spark internals. Learn why Spark exists and how to think in DataFrames — that's what actually matters in real pipelines.
A single machine (even pandas on a beefy laptop) can't hold or process data that's too large to fit in memory, or too slow to process serially. Spark solves this by distributing data and computation across many machines.
Here's a concrete way to think about it: a typical laptop has 8-16GB of RAM, and pandas usually needs several times a file's actual size in memory to work with it comfortably (every transform tends to create extra copies along the way). A 2GB CSV can already feel uncomfortable in pandas. A 200GB dataset simply can't fit on one machine, no matter how well you write the code. That's the hard line where "make pandas faster" stops being an option, and spreading the work across multiple machines becomes the only way forward.
# pandas: this either works fine on your laptop, or it doesn't fit — no middle ground
import pandas as pd
df = pd.read_csv("orders.csv") # entire file loaded into ONE process's memory
result = df.groupby("customer_id")["amount"].sum()
# PySpark: same logic, but the read AND the computation are split across many machines
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = spark.read.csv("s3://bucket/orders/", header=True) # split into partitions across the cluster
result = df.groupBy("customer_id").sum("amount") # each executor sums its own partitions firstNotice the code looks almost identical to pandas — that's on purpose, PySpark's DataFrame API was designed to feel familiar. But underneath, it works completely differently: spark.read.csv never loads the whole file into one process's memory. It's split into partitions the moment it's read, and groupBy().sum() runs as a distributed operation — each partition computes its own partial sum, and those get combined at the end — instead of one process working through the whole dataset alone.
main()/script; it plans the job and sends tasks out.Let's follow one job through this architecture, step by step, using df.groupBy("customer_id").sum("amount") as the example:
customer_id into the finished result — which only actually flows back to the driver once you call an action like .show() or .collect().A single machine can run Spark too. local[*] mode treats your own CPU cores as "executors" — this is exactly how you'll write and test PySpark code on a laptop before it ever runs on a real multi-machine cluster.
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("orders-pipeline").getOrCreate()
df = spark.read.parquet("s3://bucket/orders/")
df.printSchema()
df.show(5)
df2 = df.select("customer_id", "total_amount") \
.filter(df.total_amount > 0) \
.withColumn("amount_dollars", df.total_amount / 100)select, filter, withColumn) returns a new DataFrame rather than modifying in place.Common DataFrame operations beyond select/filter, the ones you'll reach for constantly in a real transform:
from pyspark.sql import functions as F
df = (
df.withColumn("order_date", F.to_date("order_date")) # cast a string column to a real date
.withColumnRenamed("amt", "total_amount") # rename a column
.dropna(subset=["customer_id"]) # same idea as pandas dropna
.withColumn("tier",
F.when(F.col("total_amount") >= 500, "high")
.when(F.col("total_amount") >= 100, "medium")
.otherwise("low")) # F.when/otherwise = SQL's CASE WHEN
.groupBy("customer_id")
.agg(
F.count("order_id").alias("order_count"),
F.sum("total_amount").alias("revenue"),
)
)pyspark.sql.functions — usually imported as F — is where almost every column-level operation lives: F.when, F.col, date/string functions, aggregates. You'll notice this reads almost line-for-line like the pandas transform from Week 3, and the SQL CASE expression from Week 1-2. That's on purpose — PySpark was built to feel familiar to both.
df.createOrReplaceTempView("orders")
result = spark.sql("""
SELECT customer_id, SUM(total_amount) AS revenue
FROM orders
GROUP BY customer_id
ORDER BY revenue DESC
""")
result.show()Every SQL skill from Week 1-2 carries over directly — just register a DataFrame as a temporary view, and query it with plain SQL. Underneath, Spark SQL and the DataFrame API turn into the exact same execution plan, so use whichever one is easier to read for the task at hand.
Proving they're the same plan — this is worth seeing once so "compiles to the same execution plan" isn't just an assertion to take on faith:
df.createOrReplaceTempView("orders")
sql_version = spark.sql("SELECT customer_id, SUM(total_amount) FROM orders GROUP BY customer_id")
api_version = df.groupBy("customer_id").sum("total_amount")
sql_version.explain()
api_version.explain()
# both print the exact same physical plan — Spark's Catalyst optimizer
# parses SQL into the identical internal representation the DataFrame API builds directlyIn practice, most teams default to Spark SQL for anything with several joins or CTEs — it's more readable than a long chain of DataFrame methods, and your Week 1-2 SQL knowledge carries straight over. They reach for the DataFrame API instead when Python's control flow actually helps — looping over a list of tables, building column expressions in code, or mixing in non-SQL logic.
| Transformations (lazy) | Actions (trigger execution) |
|---|---|
select, filter, withColumn, groupBy, join | show(), collect(), count(), write.parquet() |
.show() or .count().Watching laziness happen — this line runs instantly, no matter how large the underlying data is, because nothing has executed yet:
big_df = spark.read.parquet("s3://bucket/orders/") # instant — just registers where to read from
filtered = big_df.filter(big_df.total_amount > 100) # instant — just adds a step to the plan
selected = filtered.select("customer_id", "total_amount") # instant — still just a plan
selected.show() # THIS is where Spark actually reads data, applies the filter, and computes — all at onceBecause Catalyst can see the whole plan before running anything, it's free to reorder steps for you — for example, pushing your filter down so it happens during the file read itself, skipping non-matching files entirely, even though you wrote .filter() as a separate line after .read(). This "plan everything, then run it once" approach is exactly why chaining lots of small transformations in PySpark doesn't slow things down the way it would in pandas, where each line runs immediately.
df.rdd.getNumPartitions()
df = df.repartition(8) # shuffle to a new number of partitions
df = df.coalesce(2) # reduce partitions without a full shufflerepartition shuffles data to rebalance partition count/distribution; coalesce is a cheaper way to reduce partition count when you don't need a full shuffle.A concrete symptom of the wrong partition count — the Spark UI's Stages tab shows one task per partition, and this is usually how the problem actually surfaces:
# symptom: 200 tiny tasks each processing a few KB, most of the job's time
# spent on scheduling overhead rather than real work
df = spark.read.parquet("s3://bucket/small_daily_export/") # e.g. only a few MB total
df.rdd.getNumPartitions() # -> 200 (Spark's default shuffle partition count)
df = df.coalesce(4) # collapse down to a sane number for this data's actual size
# opposite symptom: a few enormous partitions, most executors sitting idle
# while one or two executors churn through a skewed partition alone
df = df.repartition(200, "customer_id") # explicitly shuffle into 200 partitions, spread by keySpark's default of 200 shuffle partitions (spark.sql.shuffle.partitions) is a reasonable guess for a large cluster, but it's often just wrong for small local jobs or small data. Adjusting this one setting is one of the easiest, highest-impact performance fixes in Spark — it's worth checking first whenever a job feels slower than the data size would suggest.
Moving data across the network between executors so related rows end up on the same partition — required for operations like groupBy, wide joins, and repartition.
Here's why groupBy specifically needs a shuffle: before Spark can add up total_amount per customer_id, every row for a given customer needs to land on the same executor — you can't get a correct total if that customer's rows are scattered across five machines that never talk to each other. A shuffle is the network step that physically moves rows around so all of customer 101's data ends up together, all of customer 102's data ends up together, and so on — before any actual summing happens.
# filter BEFORE the shuffle-triggering groupBy — shrinks what has to move across the network
df.filter(df.order_date >= "2026-01-01") \
.groupBy("customer_id") \
.sum("total_amount")
# NOT this — filtering after groupBy still shuffled the full unfiltered dataset first
df.groupBy("customer_id").sum("total_amount").filter(...) # too late, the expensive part already happenedCatalyst's optimizer often handles simple cases like this for you automatically. But understanding why "filter early" is the right instinct — smaller data before an expensive shuffle beats filtering after — is what helps you reason about performance in cases the optimizer can't fix on its own, like a shuffle triggered by joining two large tables together.
orders.join(customers, on="customer_id", how="left")
from pyspark.sql.functions import broadcast
orders.join(broadcast(small_lookup_table), on="product_id", how="left")broadcast()) when the optimizer guesses wrong.Regular join vs broadcast join, what actually moves across the network:
# regular join: BOTH orders (huge) and customers (small) get shuffled so matching
# customer_id values land on the same executor — expensive even though customers is tiny
orders.join(customers, on="customer_id")
# broadcast join: the small `customers` table is copied WHOLE to every executor;
# `orders` never moves at all — each executor joins its local orders partition
# against its own local full copy of customers, no shuffle needed
from pyspark.sql.functions import broadcast
orders.join(broadcast(customers), on="customer_id")The whole win comes down to which side has to move. A regular join pays the shuffle cost for both tables, no matter their size. A broadcast join pays a one-time cost to copy the small table to every executor — and then the large table never has to move at all. This is exactly the star-schema shape from Week 4 (a large fact table joined to small dimension tables), which is why broadcast joins show up constantly in real pipelines.
Spark's default cutoff for auto-broadcasting a table is 10MB (spark.sql.autoBroadcastJoinThreshold). Raise it if you know a bigger dimension table will still comfortably fit in executor memory, or force broadcast() yourself when Spark's own size estimate is off — which happens fairly often right after a filter Spark can't accurately predict the result size of.
When a shuffle splits data unevenly, one task ends up doing almost all the work while the rest of the cluster sits idle waiting for it to finish.
groupBy, a join on a key, or a window function partitioned by a key.A concrete example of skew: say you're grouping orders by customer_id, and one customer is actually a test account or a big retail partner with 40 million orders while every other customer has a few hundred. Spark's shuffle sends every row for a given customer_id to the same partition — so that one partition ends up holding 40 million rows, while every other partition holds a few hundred. Spark still waits for the slowest task to finish before the job is considered done, so that one oversized partition alone can dominate the entire job's runtime.
| customer_id | row count | partition |
|---|---|---|
| 1 | 340 | Partition 0 |
| 2 | 512 | Partition 1 |
| 7 (the skewed one) | 40,000,000 | Partition 4 |
| 8 | 289 | Partition 5 |
Every other executor finishes almost instantly. The one holding customer 7's partition becomes a straggler that the whole job waits on — classic data skew.
1. Salting — artificially spread a hot key across several partitions. Attach a random number to the skewed key before the shuffle, group on that combined key so the hot key's rows land on several different partitions instead of one, then combine the partial results back together afterward:
from pyspark.sql.functions import concat, lit, floor, rand
SALT_BUCKETS = 10
# stage 1: spread the hot key across N salted sub-keys, aggregate partially
salted = df.withColumn(
"salted_key",
concat(df.customer_id, lit("_"), floor(rand() * SALT_BUCKETS).cast("int"))
)
partial = salted.groupBy("salted_key", "customer_id").sum("total_amount")
# stage 2: strip the salt back off and combine the partial sums into the real total
final = partial.groupBy("customer_id").sum("sum(total_amount)")Customer 7's 40 million rows are now spread across 10 different salted sub-keys instead of piling onto one partition, so 10 executors share the work instead of 1. Stage 2's final groupBy is cheap, because by then the data has already been mostly summed down to just a handful of rows per real customer_id.
2. Isolate the skewed key and handle it separately. If you know exactly which key (or keys) are skewed — often true in practice, since it's usually the same test account or big customer every time — split your DataFrame in two and handle each path differently:
skewed_keys = [7] # known from prior runs, or found by profiling row counts per key
skewed_df = orders.filter(orders.customer_id.isin(skewed_keys))
normal_df = orders.filter(~orders.customer_id.isin(skewed_keys))
# the skewed slice is now small enough to broadcast-join or process on its own
skewed_result = skewed_df.join(broadcast(customers), on="customer_id")
normal_result = normal_df.join(customers, on="customer_id")
result = skewed_result.unionByName(normal_result)This trades a bit of extra code for a big win: the skewed key gets a code path built for its size, and the rest of the data goes through the normal, efficient path without being held back by it.
3. Let Spark handle it automatically with Adaptive Query Execution (AQE). Spark 3.x can detect skewed partitions during a shuffle (using real, observed partition sizes, not just an upfront guess) and automatically split an oversized partition into several smaller ones:
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")AQE is on by default in current Spark versions, and it's worth trying before reaching for manual salting — it solves the same problem with zero code changes to your actual transformations. Manual salting is still worth knowing for older Spark versions, for skew inside a plain groupBy (AQE's skew handling specifically targets joins), or for cases extreme enough that you want more control than the automatic version gives you.
groupBy, PARTITION BY) where you can split the work and recombine it afterward. It doesn't apply to a plain join the same way — for a skewed join, reach for AQE's skew join handling first, or the isolate-and-broadcast pattern above.df_expensive = df.filter(...).join(...).groupBy(...).agg(...)
df_expensive.cache() # or .persist()
df_expensive.count() # first action materializes and caches it
df_expensive.show() # reuses the cached result, doesn't recomputedf_expensive would recompute the entire chain from scratch..cache() stores the DataFrame in memory (spilling to disk if needed) after the first action, so subsequent actions reuse it.Seeing the cost of skipping it — without caching, this recomputes the entire filter+join chain twice, from the original data read, once per action:
expensive = orders.filter(orders.total_amount > 100).join(broadcast(customers), on="customer_id")
count = expensive.count() # runs the FULL chain: read -> filter -> join
top10 = expensive.orderBy(expensive.total_amount.desc()).limit(10).collect() # runs it ALL again from scratch# with caching, the second action reuses the already-computed result
expensive = orders.filter(orders.total_amount > 100).join(broadcast(customers), on="customer_id")
expensive.cache()
count = expensive.count() # runs the full chain once, AND stores the result in executor memory
top10 = expensive.orderBy(expensive.total_amount.desc()).limit(10).collect() # reuses the cached data — no recompute.cache() is just shorthand for .persist(StorageLevel.MEMORY_AND_DISK) — it tries memory first, and spills over to disk if the DataFrame doesn't fit, instead of just failing. Call .unpersist() once you're done reusing a cached DataFrame, so its memory gets freed up for the rest of the job instead of sitting there unused.
from pyspark.sql import Window
from pyspark.sql.functions import row_number, lag
w = Window.partitionBy("customer_id").orderBy("order_date")
df.withColumn("order_seq", row_number().over(w)) \
.withColumn("prev_amount", lag("total_amount").over(w))This is the same idea as SQL window functions from Week 1-2, giving the same result — PARTITION BY/ORDER BY map directly onto Window.partitionBy()/.orderBy(). If you're already solid on SQL window functions, this is mostly just learning the same idea's PySpark spelling.
The "latest record per customer" pattern, in PySpark — this is the direct port of the Week 1-2 SQL dedup pattern:
from pyspark.sql import Window
from pyspark.sql.functions import row_number, col
w = Window.partitionBy("customer_id").orderBy(col("updated_at").desc())
latest = (
df.withColumn("rn", row_number().over(w))
.filter(col("rn") == 1)
.drop("rn")
)Notice you can filter directly with .filter(col("rn") == 1) right after adding the column. Unlike raw SQL, there's no need to wrap it in a separate subquery — each DataFrame step is already its own step in the plan, not one giant query.
partitionBy) force all data onto one executor to compute — a common performance trap. Always partition a Spark window by something, even if it's a coarse grouping, unless the dataset is genuinely small.df = spark.read.parquet("s3://bucket/orders/")
result = (
df.groupBy("customer_id")
.sum("amount")
)
result.show()Putting several concepts from this page together — a small end-to-end job: read from S3, broadcast-join to a dimension table, dedupe with a window function, and write back out:
from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import broadcast, col, row_number, sum as spark_sum
spark = SparkSession.builder.appName("orders-summary").getOrCreate()
orders = spark.read.parquet("s3://bucket/raw/orders/")
customers = spark.read.parquet("s3://bucket/raw/customers/") # small — good broadcast candidate
# dedupe: keep only the latest version of each order
w = Window.partitionBy("order_id").orderBy(col("updated_at").desc())
orders_clean = (
orders.withColumn("rn", row_number().over(w))
.filter(col("rn") == 1)
.drop("rn")
)
# broadcast join + aggregate
result = (
orders_clean.join(broadcast(customers), on="customer_id")
.groupBy("customer_id", "city")
.agg(spark_sum("total_amount").alias("revenue"))
)
result.write.mode("overwrite").parquet("s3://bucket/curated/revenue_by_customer/")Every line here maps back to something covered above: partitioned S3 reads, a window-function dedupe, a broadcast join to avoid shuffling the large table, and a final write — the same shape as the Glue job in Week 5, and the "PySpark / dbt" box in your capstone architecture.