Distributed Batch Processing with PySpark

Spark spreads a DataFrame across a cluster as partitions — chunks of rows. A driver program plans the work; executors run it. Each executor has some cores, and each core processes one partition at a time as a task.

So the basic arithmetic is:

• 1 partition → 1 task.
• Tasks run in waves of (total executor cores): waves = ⌈partitions ÷ cores⌉.
• Stage time ≈ waves × time per task.

20 partitions on 8 cores need 3 waves — and the last wave uses only 4 cores while 4 sit idle. Too few partitions leaves cores idle; far too many creates scheduling overhead. A common rule of thumb is 2–4 partitions per core.

Transformations (filter, select, withColumn, groupBy, join) don't execute — they add steps to a logical plan. Only an action (count, collect, show, write) triggers execution.

Laziness lets Spark's Catalyst optimiser rewrite your whole plan before running it: push filters down to the data source, drop unused columns, reorder joins, then the Tungsten engine generates efficient code. df.explain() shows the physical plan — look for Exchange (a shuffle) and PushedFilters.

The flip side: every action re-runs the plan from the source unless you cache. Two count()s on the same DataFrame read the data twice.

• Narrow transformations (filter, select, withColumn, map) work inside each partition. No data moves between machines.
• Wide transformations (groupBy, join, orderBy, distinct) need rows with the same key to end up together. Spark shuffles: hashes each row's key, writes partition files to disk, and sends them across the network. That's the Exchange in the plan and usually the most expensive part of a job.

Spark reduces shuffle cost with partial aggregation: each partition first computes its own partial sums per key, so at most partitions × distinct keys rows are shuffled instead of every row. That's why groupBy("commodity").avg("price") on 10 million rows shuffles only a few thousand.

Spark's fixed costs — starting a JVM, planning, serialising tasks, scheduling, shuffles — are seconds, not milliseconds. On small data they dominate. In Lab 8 on 10,000 rows, DuckDB took 1.39 s and Spark 11.21 s.

.cache() also isn't free: the first action has to materialise the cached data in memory. It pays off only when the cached DataFrame is reused by enough later actions. In our benchmark, caching a DataFrame used by two actions made the script slower (8.85 s vs 3.04 s).

Rule of thumb: if the data fits comfortably on one machine, use a single-node engine (DuckDB, Polars, pandas). Reach for Spark when data or compute genuinely exceeds one machine.