My working days at the moment are a Windows agent for Jenkins and a build that signs things, which is about as far from distributed data processing as it is possible to get while still using a computer. So this is a weekend post, and it is a beginner's post. I have read about Apache Spark for years in the way you read about a city you have not visited. Until this month I had never written a job in it.

What finally moved me was Amazon EMR Serverless, announced in preview at re:Invent at the end of November. Its pitch is that you stop provisioning a cluster. I have watched enough people spend a fortnight on node counts and instance families to find that genuinely interesting, and it raised the question I actually wanted answered: if the cluster stops being my problem, what is left that still is?

What the service takes off the table

The model has two nouns. An application is the long-lived thing you create, and it pins two choices: the open-source framework, and the Amazon EMR release that fixes its version. A job run is a request you submit to that application — a PySpark script, a Hive query — which it executes asynchronously and tracks to completion. You can have many job runs against one application, running concurrently, each with its own runtime IAM role for reaching S3.

Underneath, the application uses workers. EMR Serverless works out what a job needs, provisions workers, scales them up and down through the stages of the job, and decommissions them at the end. There is an optional pre-initialized capacity setting that keeps a warm pool ready so jobs start in seconds rather than waiting for provisioning, which is aimed squarely at the iterative case where you are running the same job over and over with small changes.

Set against EMR on EC2, the list of questions that disappears is not short. How many core nodes. Which instance family. Task nodes on spot, and what happens when they go away. Whether the cluster is long-lived or transient, and if transient, what terminates it when a job hangs. None of those have good answers available in advance, and all of them are things I would rather not become good at.

In January this is still a preview: you put your name on a form, it runs in N. Virginia only, and the frameworks are Spark 3.1.2 and Hive 2.0. I signed up and have been waiting, which turned out to be the useful part of the month. It sent me to learn the thing the service is actually running.

So I pinned a local Spark to the version in the preview

There is no point learning Spark 3.2 semantics for a preview that runs 3.1.2, so I matched it and kept the whole loop on my laptop:

docker run --rm -it \
  -v "$PWD":/work -w /work \
  -p 4040:4040 \
  apache/spark-py:v3.1.2 /opt/spark/bin/pyspark

Port 4040 is the part I would tell a past version of myself to open first. The Spark UI while a job is running is the only reason any of the rest of this post makes sense to me — jobs, stages, tasks, and the time each one took, which is a far better teacher than the documentation.

The dataset is deliberately dull: a few hundred megabytes of CSV that I already had, with a category column and an amount column.

from pyspark.sql import functions as F

events = spark.read.csv("/work/data/", header=True, inferSchema=True)

recent = events.filter(F.col("amount") > 100).select("category", "amount")
totals = recent.groupBy("category").agg(F.sum("amount").alias("total"))

totals.show()

Six lines, and three things in them that I had wrong.

Nothing runs until something has to come back

The filter and the select do not filter or select anything. They build a plan. Spark separates transformations, which are lazy and only describe work, from actions, which are the things that force it — show, count, collect, writing to storage. Until totals.show(), nothing has read a byte.

This sounds like trivia and it is not, because it means the thing you time is almost never the thing that is slow. My first instinct was to measure line by line, and every line took no time at all until one of them took all of it.

The useful habit is asking for the plan rather than guessing at it:

totals.explain()

The physical plan it prints is where the Exchange steps show up, and those are the ones that cost. Which brings up the second thing.

The partition is the unit of parallelism, not the machine

I had a vague picture of Spark spreading work across machines. That is wrong at exactly the point where it matters. Work is spread across partitions, and a task processes one partition. How many partitions you have is decided by how your data is laid out, not by how many workers you asked for.

events.rdd.getNumPartitions()

The consequence arrives immediately with compressed files. A gzipped CSV cannot be split — nothing can start reading it from the middle — so however large it is, it is one partition, and it is one task on one core while everything else in your expensive cluster watches. My first run used gzipped input and was perfectly, uniformly slow no matter what I gave it. Snappy-compressed Parquet does not have that problem, and the fix was a format change rather than anything to do with capacity.

Ten thousand tiny files are the same mistake wearing a different hat: thousands of partitions, each with real scheduling overhead, doing almost nothing each.

Some operations have to move data, and those are the ones that hurt

The distinction that made everything else fall into place is narrow against wide. A narrow transformation — filter, select, a per-row function — can be done inside a partition, no coordination needed. A wide one — groupBy, join, distinct, repartition — needs rows that share a key to end up in the same place, and the rows are not in the same place. So they get written out, moved across the network, and read back in. That is a shuffle, and it is a stage boundary.

My six-line job has exactly one, at the groupBy. Everything before it is free in comparison. Once I started reading a plan looking for Exchange first and ignoring the rest, the Spark UI stopped being a wall of numbers and started being a short list of things worth caring about.

The number nobody chose: 200

Here is the one that made the whole exercise worth writing down.

When Spark shuffles, the number of partitions it shuffles into comes from spark.sql.shuffle.partitions, and the default is 200. Not 200 because of my data, or my cluster, or my job. Just 200.

On the few hundred megabytes I was playing with, that is 200 tasks, most of them processing a trivial amount of data, each paying scheduling cost, and 200 output files if I write the result. My aggregation over a small dataset spent most of its wall-clock time on the overhead of pretending to be big. Setting it sensibly:

spark.conf.set("spark.sql.shuffle.partitions", 8)

took the job from tens of seconds to a few. On a genuinely large dataset the same default is the opposite mistake, and 200 is far too few.

I want to be careful about the size of the claim here: 8 is right for my laptop and my toy file and nothing else, and the real answer depends on data volume and available cores. The point is not the number. The point is that this is a knob with a fixed default that has no relationship to the problem, and it sits directly on the critical path of most jobs anyone writes.

What serverless automates, and what it does not

EMR Serverless scales workers to the parallelism a job asks for at each stage. That is real and it is useful. What it cannot do is change what the job asks for. It will not choose a shuffle partition count for me, notice that one key holds forty per cent of the rows, pick a different join strategy, or object that I have pointed it at ten thousand small files.

And there is a sting in that which I had not thought about before this month. On a cluster I have already paid for, a badly partitioned job wastes capacity that was sitting there anyway; the cost is my afternoon. On a service that provisions and meters resources per job, the same inefficiency converts directly into a bill. Two hundred near-empty tasks occupy real workers for real seconds. Serverless does not make Spark tuning optional. It moves it from a capacity problem, where the waste is invisible, to a cost problem, where it is itemised.

That seems like the honest summary of the trade. I get to stop having opinions about instance families, and in exchange my ignorance of partitioning acquires a price.

Where I have actually got to

Not far, and I would rather say so than imply otherwise. I can write a PySpark job, read its plan, find the shuffle, and explain why it is there. I have no production experience of Spark at all, and everything above is a fortnight of weekends against a small file on one machine — which is precisely the setting in which distributed systems tell you the least.

The plan when the preview opens up is unglamorous: take one job that means something, run it on a small EMR cluster and on EMR Serverless with the same input, and compare the wall-clock and the bill rather than the adjectives. Use pre-initialized capacity for the iteration loop, because waiting on provisioning between attempts is how people stop iterating. And keep the local 3.1.2 container for learning, because the fastest way to understand a shuffle is still to cause one on a laptop and watch it happen on port 4040.