Sunday's post ended with a Spark container on my laptop reading a folder of CSV files off local disk. The obvious next step was to read the same data out of S3 instead, because that is where data actually lives and because it is the shape EMR Serverless will want when the preview lets me in.

That took an evening, and not for any of the reasons I had budgeted for. Every single thing that went wrong was called org.apache.hadoop something — in a job with no Hadoop cluster, no HDFS, and no MapReduce anywhere near it. Which turned out to be the most useful thing I learned this month, so here it is written down.

"Hadoop" is three things, and Spark replaced one of them

I had been carrying around the sentence "Spark replaced Hadoop" without ever asking what the words meant. Hadoop is not a program. It is a project containing three separable pieces:

• HDFS, a distributed filesystem — the storage.
• YARN, a cluster resource manager — the thing that decides which machine runs what.
• MapReduce, a compute engine — the thing that actually processes data.

Spark replaced exactly one of those. It is a compute engine. It stores nothing, and on its own it schedules nothing across a cluster.

So "we moved off Hadoop onto Spark" can mean at least four different things, and it is worth asking which. Usually it means the MapReduce jobs became Spark jobs and everything underneath stayed where it was.

The difference between the two engines is narrower than the marketing suggested and lines up neatly with Sunday's post. MapReduce materialises intermediate results to disk between every map and reduce step. Spark builds a plan of stages and keeps data in memory across the narrow transformations inside a stage, writing out only at shuffle boundaries. That is the whole advantage — which means Spark's lead over MapReduce is largest on jobs with many chained steps and smallest on a job that is essentially one big shuffle.

What Spark still needs from the other two

Storage has to come from somewhere. HDFS is one option, and on AWS almost nobody chooses it as the store of record: you put the data in S3 and let compute be a separate, disposable thing. That separation is the actual architectural change of the last decade, and it is bigger than the engine swap.

Scheduling has to come from somewhere too. Spark can run standalone, on YARN, on Mesos, or on Kubernetes. On EMR it runs on YARN. On EMR Serverless you never see a resource manager at all, which is precisely what Sunday's post was about.

So a modern Spark-on-AWS job uses none of HDFS, none of MapReduce, and — on the serverless option — none of YARN that you can observe. Zero of the three. And the jars are still everywhere.

The fourth thing, which is the one that bit me

The piece nobody lists is Hadoop Common: the shared libraries. And the important one there is org.apache.hadoop.fs.FileSystem, the abstraction through which everything in this ecosystem reads and writes.

Spark does not have an S3 client. It asks Hadoop's FileSystem for a path, and the implementation for s3a:// lives in a Hadoop module called hadoop-aws. That is the entire explanation for why my stack traces looked the way they did.

The binary I had been using is spark-3.1.2-bin-hadoop3.2, which ships Hadoop 3.2.0 jars — but not the cloud connectors. So the first attempt ends at:

java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found

The fix is to add the module, and the version is not a matter of taste:

pyspark \
  --packages org.apache.hadoop:hadoop-aws:3.2.0,com.amazonaws:aws-java-sdk-bundle:1.11.375

3.2.0 because that is the version of the Hadoop jars already inside the tarball, and hadoop-aws is not independently versioned from the rest of Hadoop. Reaching for a newer 3.3.x hadoop-aws next to 3.2.0 hadoop-common does not fail cleanly at startup; it fails later with a NoSuchMethodError from somewhere unrelated-looking. 1.11.375 is not a choice either — it is the AWS SDK version that hadoop-aws 3.2.0 was compiled against and declares as its dependency.

That is two version numbers that must agree with a third one you did not pick, and nothing checks any of it for you. An hour went here.

Then the configuration, which is Hadoop configuration wearing a Spark prefix:

hadoop_conf = spark.sparkContext._jsc.hadoopConfiguration()
hadoop_conf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
hadoop_conf.set(
    "fs.s3a.aws.credentials.provider",
    "com.amazonaws.auth.DefaultAWSCredentialsProviderChain",
)

events = spark.read.parquet("s3a://my-bucket/events/")

The credentials provider is worth setting explicitly rather than putting an access key into Spark configuration, where it will end up in the event log, the UI, and eventually somebody's screenshot. The default chain picks up the environment, the profile, or the instance role, which is what you want in all three of the places this job might run.

Reading was fine. Writing was the interesting problem.

Reading worked after that. Writing a partitioned Parquet output took far longer than the computation that produced it, and sat there doing nothing visible at the end of an otherwise finished job.

This is not a bug and it is not a misconfiguration. It falls straight out of the first half of this post.

Hadoop's commit protocol was designed for HDFS, where renaming a directory is a cheap atomic metadata operation. FileOutputCommitter uses that: each task writes its output into a temporary location, and committing means renaming it into the final path. Rename is how you get the guarantee that either the whole output appears or none of it does.

S3 has no rename. The S3A client emulates one by copying every object to the new key and deleting the original. So the commit phase of the job is a full copy of everything you just produced, performed after all the real work is done — and it is not atomic either, so it can fail halfway and leave a partial result behind, with nothing preventing two processes attempting it at once.

The answers exist and both work the same way. Since Hadoop 3.1 there are the S3A committers — directory, partitioned and magic — which use S3 multipart uploads: the task streams its data to S3 as it works but does not complete the upload, so committing is a cheap API call that makes already-uploaded bytes visible, instead of a copy. On EMR, the EMRFS S3-optimized committer does the same thing for s3:// paths, and AWS's own comparison against FileOutputCommitter v1 puts the difference in multiples rather than percentages.

Turning the committers on in open-source Spark needs the binding classes from Spark's hadoop-cloud module, which is not in the binary tarball either. That is where I stopped for the evening, so I am reporting the shape of the fix rather than a measurement of it — I have not yet run a job with a committer configured, and I would want to see the numbers on my own data before repeating anybody's.

The consistency problem that already went away

One thing worth separating, because a lot of what I read while digging conflated them.

S3 used to be eventually consistent for listings. A job could write files and then genuinely not see them, which is why S3Guard and EMRFS Consistent View existed — a DynamoDB table whose entire purpose was recording which objects really exist. On 1 December 2020 S3 became strongly read-after-write consistent for all requests, at no extra cost and with no change to performance, and that whole layer became redundant.

The rename problem survived that, because it was never a consistency problem. It is a cost problem: rename is a metadata update on a filesystem and a full copy on an object store. Strong consistency makes the copy correct. It does not make it cheap. If you read older material about writing to S3 from Spark, half the advice has expired and half of it has not, and this is the line that separates them.

What I will actually do differently

The practical change is small and I suspect it will save me a lot of time: read the package name before reading the message.

org.apache.spark in a stack trace is my logic, my plan, or my partitioning — Sunday's post. org.apache.hadoop.fs is the storage layer, and the answer is almost always a version number, a credentials provider, or a committer. I had been treating those as one undifferentiated category called "Spark is broken", and they are not remotely the same problem.

The broader thing I got wrong is more interesting. The three famous pieces of Hadoop are the ones every migration story is about, and all three are genuinely gone from a job like mine. The piece nobody migrates off is the unglamorous one in the middle — the filesystem API — because it is the interface every connector in the ecosystem was written against. EMR Serverless takes away the cluster and the resource manager, which is real and which I want. It does not take away hadoop-aws, and the assumptions baked into that code are from a world where rename was free.