Notes From an Advanced Apache Spark Training

Why It’s Faster Than MapReduce

The first part is a tour of the ecosystem, and the useful bit is the argument for Spark. In MapReduce, a real analysis is a chain of jobs, and each job writes its result back to HDFS so the next one can read it. His example is k-means with Mahout, which needs about 17 MapReduce jobs even for 5 MB of data. Spark keeps the data in memory between steps, so one Spark job can be “20 maps and 15 reducers and a whole bunch of joins”.

He backs this with the speeds he uses for the whole class:

Spark doesn’t replace Hadoop, by the way. Hadoop is three projects (HDFS, YARN and MapReduce), and Spark replaces only the MapReduce engine. It reads almost anything through Hadoop’s InputFormat, which is why all of Spark’s I/O code was only ~400 lines for a long time.

Fun fact from this part: Spark started as a sample app for Mesos. Matei Zaharia wrote it to “spark” an ecosystem of apps on top of Mesos, and it outgrew Mesos.

He also recommends a few resources here that I’m adding to my list. The Learning Spark book (he says it has information you won’t find anywhere else), the Apache Spark user mailing list (most questions get answered within hours), Spark Packages for anything that isn’t in Spark itself, and the 2012 RDD paper.

A Partition Is a Task Is a Thread

The RDD section has the most useful idea in the whole class, and it’s also the simplest. An RDD is split into partitions, each partition needs one task, and each task is one thread in an executor. So an RDD with 5 partitions can use 5 cores and that’s it, even if your cluster has 1,000 of them. His rule of thumb is that 1,000 to 10,000 partitions in an RDD is perfectly normal, and a 100-node cluster needs at least 100.

You make an RDD by reading from storage or by calling parallelize on a collection in the driver. parallelize needs the whole dataset in the driver’s memory, so it’s only for prototypes and tests.

When you read a file from HDFS, you get one partition per block. A filter keeps the same number of partitions as its parent, even when some of them end up empty. This is why the usual pattern is to read, filter, and then coalesce or repartition to something that fits your cluster.

None of this runs until you call an action. Transformations are lazy, so reading and filtering only build a DAG of metadata. Spark doesn’t even check that the input file exists until you call count or collect.

And be careful with collect, because it pulls every partition into a driver that probably has a gig or two of heap. On anything big, save it or sample it first.

Caching is lazy too. Calling cache() only marks the RDD, and it’s stored the first time an action materializes it. The beginner mistake he calls out is caching the base RDD right after reading it. Do your filters and maps first, and cache the cleaned RDD that several actions actually reuse. A production app has hundreds of RDDs, so be picky.

It also doesn’t have to fit in memory. If only 30 of 100 partitions fit, 30 are cached and the other 70 get recomputed when they’re needed, so performance degrades instead of falling over.

Every RDD is defined by five things. It has a list of partitions, its dependencies on parent RDDs, a function to compute one partition from the parents, optional preferred locations (for example, where an HDFS block lives), and an optional partitioner for key/value data. That’s the whole interface, and it’s what makes recovery work.

A HadoopRDD has one partition per block and no parents. A filtered RDD has its parent’s partitions and a one-to-one dependency on it. A joined RDD has a shuffle dependency on both parents and a HashPartitioner. If you lose a partition, Spark recomputes just that partition from this information.

He also wants you to know the type of each RDD in your DAG, even if you have to draw it on paper at first. A ShuffledRDD tells you a shuffle happened right before it, and shuffles are the expensive part.

A few more tips from this part that I really liked:

• If the docs describe a transformation in two sentences, go read the Scala source. It often has a full paragraph of comments on the same thing.
• If you write to an external database, don’t open a connection per element. Use foreachPartition or mapPartitions, open one connection per partition, push everything through it, and close it.
• Some connectors push work down to the source. The Cassandra connector, for example, can select only the columns you need and push a where filter into Cassandra, instead of reading the whole table into Spark and filtering there.
• With Cassandra, spark.cassandra.input.split.size sets the rows per partition (100,000 by default). With very wide rows, lower it, for example to 2,000. He says getting this right is “more of an art than a science”.

Four Ways to Run Spark

He opens the cluster section by saying this was the hardest thing for him to figure out when he joined Databricks. Spark runs in local mode, Standalone, YARN or Mesos, and the architecture is different in each one.

What helped me was his summary at the end. There’s always a driver and executors, and tasks always run in executors. The only thing that changes is who decides where the executors go and who starts them.

Compared to MapReduce, parallelism in Spark comes from threads and not processes. MapReduce starts a JVM per map or reduce task and has fixed map and reduce slots, so the reduce slots sit idle during a long map phase. Spark usually runs one executor JVM per machine with generic slots that can run any task. Handing a task to a slot is a thread handoff, not a JVM that takes a second to start.

The scheduler that matters most is the one in the driver, not the cluster manager. The cluster manager places executors. The driver places every task, and it tries to put each one on the executor that already holds its partition.

Where the Memory Goes

His sizing rules are to give Spark at most ~75% of a machine’s memory, give each executor at least 8 GB, and keep a heap under about 40 to 45 GB on a normal JVM or the GC pauses get bad. The 75% isn’t arbitrary. If HDFS or Cassandra runs on the same machine, they need memory too, and giving executors everything leads to swapping.

cache() is just persist(MEMORY_ONLY), which stores deserialized objects and drops partitions that don’t fit. MEMORY_ONLY_SER stores one byte array per partition, which is smaller and easier on the GC but costs CPU. MEMORY_AND_DISK spills what doesn’t fit into the local dirs, always serialized. DISK_ONLY does what it says, the _2 levels keep two copies on two executors, and OFF_HEAP stored data outside the JVM (in Tachyon at the time).

His advice for picking one:

• Use MEMORY_ONLY if it fits, then try MEMORY_ONLY_SER with a fast serializer.
• Only spill to disk if the partitions are expensive to compute, since recomputing is often as fast as reading from disk.
• Use the replicated levels sparingly. Spark isn’t HDFS, and lineage is the backup.

The cache is LRU and you can’t pin anything, so caching something huge evicts everything older. Use unpersist() when you’re done with an RDD. If an executor dies, its cache is gone and nothing rebuilds it in the background. Partitions only come back when something needs them again. In PySpark, cached data is always pickled anyway, so _SER or not makes no difference there.

The thing in this section that surprised me most: shuffle output stays on disk even if you never cache anything. Run the same action twice on a long DAG, and the second run only goes back to the last shuffle and reuses the map output files still sitting in the local dirs. That explains why a re-run can be so much faster, and why the UI shows “skipped” stages.

He also shows how the executor heap was split. It was 60% for cached RDDs with spark.storage.memoryFraction, 20% for shuffle buffers with spark.shuffle.memoryFraction, and the remaining 20% for your own code.²

If executors are getting killed for memory, figure out which of those three caused it. Running out during a shuffle means giving shuffle more. Running out in your own code means shrinking the other two, because there’s no setting for user memory itself. If you barely cache anything, he says you can drop the cache fraction to 10 or 20%, but not to zero. To see how much memory an RDD takes, look at the Storage tab, or at the driver log, which has a line like Added rdd_0_1 in memory on mbk.local:50311 (size: 717.5 KB, free: 332.3 MB) for every cached partition.

For serialization and GC, the default Java serializer is slow and it’s used for shuffles, spills, serialized caching, broadcasts, etc. So switch to Kryo in production (spark.serializer=org.apache.spark.serializer.KryoSerializer), register your classes, and raise the Kryo buffer if your objects are large.

GC cost depends on the number of objects and not their size, which makes an array of ints a lot cheaper than a linked list. You can measure it by starting executors with -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps. The parallel GC is fine for batch, but for streaming he says to use CMS or G1 to avoid long stop-the-world pauses.

Jobs, Stages and Tasks

He calls this section the most important one for writing fast Spark code, and I agree. An action triggers a job. The DAG scheduler in the driver cuts the DAG into stages and hands each stage to the task scheduler, which sends individual tasks to executors and doesn’t know anything about stages at all. If you want to go deeper, he recommends Matei Zaharia’s “Spark Internals” talk on YouTube.

Independent stages can run at the same time, but a stage that needs both of them waits for the slower one. So if one parent stage takes 3 minutes and the other takes 17, move work into the fast one if you can.

Where the stages get cut comes down to narrow and wide dependencies. A dependency is narrow when each parent partition feeds at most one child partition. That covers map, filter, union, and a join of two RDDs that are already partitioned the same way. It’s wide when one parent partition feeds many children, for example groupByKey or a join of RDDs that aren’t co-partitioned. Wide means shuffle.

Narrow steps get pipelined, so one thread reads a block, maps it and filters it all within one stage. They’re also cheap to recover, because losing a partition means recomputing one parent partition. After a wide step, losing a partition might mean rereading the whole input. This is why he suggests caching the RDD right after an expensive shuffle. And if you do cache it, it’s worth using MEMORY_AND_DISK, so an evicted partition comes back from local disk instead of from the whole input.

Another one from this part: if a transformation is going to shuffle anyway, pass numPartitions to it. Otherwise the child inherits its parent’s partition count, which may not be what you want after the shuffle.

One might ask why the famous diagram in the RDD paper has no stage boundary at the groupBy right before the join. He asked Matei the same thing (“did you make a massive mistake in your white paper?”), and the answer is that the shuffled RDD is never really materialized. The join reads the shuffle’s map output files directly.

He also does a quiz from a Cloudera blog post by Sandy Ryza. How many stages does this need?

sc.textFile("someFile.txt") \
    .map(map_func) \
    .flatMap(flat_map_func) \
    .filter(filter_func) \
    .count()

Just one, since every step is narrow. Add a groupByKey and a join and you get three.

toDebugString prints the lineage, and each level of indentation is a stage boundary. This is what the lineage of a PySpark word count looks like on Spark 3, with the call sites trimmed:

(4) PythonRDD[9] at RDD at PythonRDD.scala:53
 |  MapPartitionsRDD[8] at mapPartitions at PythonRDD.scala:145
 |  ShuffledRDD[7] at partitionBy at ...
 +-(4) PairwiseRDD[6] at reduceByKey at <stdin>:3
    |  PythonRDD[5] at reduceByKey at <stdin>:3
    |  log.txt MapPartitionsRDD[1] at textFile at ...
    |  log.txt HadoopRDD[0] at textFile at ...

The HadoopRDD followed by a map is something you get for free when reading text. That map converts Hadoop’s Writable types into normal ones.

He has a few rules for spotting a shuffle:

• repartition, join, cogroup and anything named *By or *ByKey probably shuffle.
• Passing numPartitions usually means a shuffle too, since distinct is a reduceByKey inside.
• groupByKey and friends call combineByKey, which shuffles.
• repartition(n) is literally coalesce(n, shuffle=True). So use coalesce to reduce partitions. A repartition to the same number is still useful, since it rebalances skewed partitions.

And if you only change the values of a key/value RDD, tell Spark the keys are untouched. Otherwise it drops the partitioner and the next join shuffles again. He shows this as map(..., preservesPartitioning=True).³

Someone asked how many transformations an app should have. His answer is that a few thousand is fine. Unlike MapReduce, there’s no reason to split a pipeline into many apps, so keep it as one Spark app.

Salting a Skewed Shuffle

This part isn’t in the talk. It’s the technique I reach for when a shuffle is unbalanced, and the class finally gave me the words for why it works, so I want it in the same place as everything else.

Here’s the problem. A shuffle sends every record with the same key to the same partition, because that’s what a HashPartitioner does. One partition is one task, and one task is one thread. So if one key has 40% of the records, one thread gets 40% of the work, and the stage can’t finish until that one task does. The Spark UI makes this obvious once you know to look. In the Stages tab, the max task duration is minutes while the median is seconds, and one task shows a huge shuffle read. Adding executors doesn’t help at all, since the extra cores just sit there waiting for that one thread.

Before fixing anything, I check which keys are hot:

from pyspark.sql import functions as F

(
    df.groupBy("user_id")
    .count()
    .orderBy(F.desc("count"))
    .show(10)
)

Salting splits a hot key into several keys. You add a random number from 0 to N - 1 (the salt) to the key, so user_42 becomes (user_42, 0) through (user_42, 7), and the hash partitioner spreads those 8 keys over up to 8 tasks. Then you undo the salt at the end. How you undo it depends on what you’re doing.

For an aggregation, you aggregate twice. First by the salted key, which runs in parallel across the salts, and then again by the real key, which only has N partial results per key left to combine:

import random

SALTS = 8


def add_salt(kv):
    key, value = kv
    return (key, random.randrange(SALTS)), value


partial = pairs.map(add_salt).reduceByKey(lambda a, b: a + b)
totals = (
    partial.map(lambda kv: (kv[0][0], kv[1]))
    .reduceByKey(lambda a, b: a + b)
)

This only works for aggregations you can combine in pieces, such as sums, counts, min, max, etc. A median doesn’t split like this. And to be fair to reduceByKey, it already combines on the map side before the shuffle, so a hot key in a plain sum hurts less than you’d think. Salting pays off most with groupByKey-style work and with joins, where nothing gets combined before the shuffle.

For a join, the salt has to match on both sides. You salt the big, skewed side randomly, and you copy each row of the other side once per salt value, so every salted key still finds its match:

from pyspark.sql import functions as F

SALTS = 8

events_salted = events.withColumn(
    "salt", (F.rand() * SALTS).cast("int")
)
salts = F.array([F.lit(i) for i in range(SALTS)])
users_salted = users.withColumn("salt", F.explode(salts))

joined = events_salted.join(
    users_salted, ["user_id", "salt"]
).drop("salt")

The cost is that the small side is now N times bigger and goes through the shuffle too. So I keep N small (start around the skew ratio of the hottest key to an average key, then measure), and when only a few keys are hot, I salt only those keys and give every other row a salt of 0. If the small side fits in memory, skip all of this and broadcast it instead, like the talk shows in the next section. No shuffle means no skew.

On Spark 3, it’s also worth trying adaptive query execution before salting by hand. With spark.sql.adaptive.enabled on, Spark can split skewed partitions of a sort-merge join on its own.⁴ It only covers joins, though, so I still salt aggregations myself.

Broadcasts and Accumulators

Without a broadcast, anything your closure uses is shipped with every task. A broadcast variable is sent once per executor and shared by its tasks, which is great for lookup tables, for a big feature vector in ML, or for avoiding a shuffle join when one side is small.

The implementation is fun. Broadcasts used to be plain HTTP from the driver, which made the driver’s network card the bottleneck. Now they use a BitTorrent-like protocol where the value is cut into ~4 MB chunks and executors fetch chunks from each other.

Accumulators are basically Hadoop counters, for things like counting blank lines or corrupt records. Tasks can only add to them, only the driver can read the value, and you can watch them in the Spark UI while a job runs. He says the Learning Spark book has a whole chapter on their edge cases.

lookup = sc.broadcast({"Error": 3, "Warn": 2, "Info": 1})
levels = lines.map(lambda line: line.split(",")[0])
severity = levels.map(lambda level: lookup.value[level]).sum()

blank_lines = sc.accumulator(0)

def count_blank(line):
    global blank_lines
    if not line.strip():
        blank_lines += 1

lines.foreach(count_blank)
print(blank_lines.value)

PySpark Is Mostly a JVM Program

The PySpark section changed how I think about PySpark the most. PySpark runs normal CPython, so any library with C extensions (NumPy, NLTK, etc.) just works. The SparkContext you get in Python is only a controller. It starts a normal JVM driver and talks to it over Py4J, which is fine for commands but slow for data.

Each executor JVM then starts daemon.py, which forks one Python worker per task slot. Data moves between the executor and those workers over a custom high-throughput pipe.

Reading and writing data happens in the JVM, so PySpark reuses all of Spark’s I/O code instead of reimplementing every source in Python. That’s how the first version of PySpark was only a few thousand lines. Your functions are pickled with cloudpickle, sent through the driver JVM to the executors, unpickled in the Python workers, and run there. Even collect avoids Py4J. The driver JVM writes the results to local files and Python reads them back, and big broadcasts go the same way.

The practical takeaway is that your own Python code runs at Python speed, but MLlib, shuffles, Spark SQL and DataFrames run inside the JVM. A DataFrame job from Python is about as fast as one from Scala.

Two settings from this part are worth knowing. You can switch the interpreter to PyPy with PYSPARK_PYTHON=pypy and PYSPARK_DRIVER_PYTHON=pypy, with speedups from 1.05x on a sort to 48x on a stats job, as long as you don’t need C extensions. And each Python worker gets 512 MB by default, which you can raise with spark.python.worker.memory.

Sorting 100 TB on 206 Machines

The 100 TB sort is his favorite part, and it’s easy to see why. Sorting is the purest test of the shuffle, since sorting 100 TB means ~500 TB of disk I/O and ~200 TB over the network. The rules also said the data starts on disk, so no caching was allowed at all.

That’s the same data sorted 3x faster on 10x fewer machines, in the public cloud. Spark tied for first with TritonSort from UC San Diego, a system built to do nothing but sort.

The setup was 206 i2.8xlarge EC2 machines with one big executor each, 32 slots, 8 SSDs and about 9.5 Gb/s between two random nodes. They used HDFS short-circuit reads, turned off speculative execution and compression, set locality wait to infinite, set spark.shuffle.spill=false so reducers kept everything in memory, and used sun.misc.Unsafe to keep data off the heap so GC never ran. HDFS blocks were 3.6 GB, which made 28,000 blocks, so the maps and the reduces each ran in 5 waves. He’s very clear that this is a benchmark setup and not something to copy into production.

The engineering behind it was three changes. The first is the new shuffle. The old hash-based shuffle had each map task keep one open file per reducer and write into all of them, which falls apart around 10,000 reducers because of the per-file buffers. The new sort-based shuffle (SPARK-2045) sorts a bit of the input at a time, writes one file at a time, and merge-sorts them into a single file indexed by reducer. It’s been tested with 250,000 reducers. He adds that for a small number of partitions the hash-based one might still be a bit faster, since it skips the merge.⁵

The second is the Netty transport (SPARK-2468), which sends map output straight from the file to the network card instead of copying it through a kernel buffer and the executor first. The third is the external shuffle service (SPARK-3796). The Worker or Node Manager serves map output instead of the executor that wrote it, which takes load off busy executors. It’s also required for dynamic allocation, since otherwise a removed executor’s shuffle files would disappear before the reducers fetched them.

The result saturated everything. The map phase hit ~3 GB/s of disk per node and the reduce phase hit ~1.1 GB/s of network per node, which is the whole 10 Gb link. On the reducer side, there’s also a limit on how much data each reducer pulls at once (48 MB by default) so it doesn’t flood its own network card.

Streaming in Small Batches

The class ends with Spark Streaming.⁶ A DStream is a sequence of RDDs, one per batch interval, and transformations you apply to the DStream run on every RDD it produces. The point is that you write batch and streaming code with the same API instead of MapReduce plus a separate Storm codebase. In this version, the Python API couldn’t read from Kafka or Flume yet, only files and sockets.

Batches can be as small as ~500 ms. He says that covers 90 to 95% of real-time use cases, and the rest (high-frequency trading, for example) need something like Storm.

A receiver is a long-running task that permanently takes one slot, which is why a streaming app in local mode needs at least local[2]. The receiver cuts incoming data into a block every 200 ms and replicates each block to a second executor. So with a 600 ms batch interval, each batch’s RDD has 3 partitions. The 200 ms block interval is configurable, and it’s effectively your partition size, but he says Tathagata Das (who wrote Spark Streaming) told him you probably don’t need to touch it. local[2] is also only the minimum. In general, you want way more threads than that.

The rule that matters is that processing time has to stay below the batch interval, or you fall further behind with every batch. The easy fix is a smaller batch interval with less data per batch, down to about 500 ms. You can also repartition early to get more parallelism. And one receiver is one thread, so to ingest more you run several receivers, one per Kafka topic for example, and union the DStreams.

The streaming tab in the UI shows processing time percentiles over the last 100 batches, the input rate, and each receiver, which is where you’d see that you’re falling behind.

He finishes with the API. transform gives you any RDD operation that DStreams don’t expose, for example joining each batch with a static RDD of spam data. updateStateByKey keeps state across batches and needs a checkpoint directory. Windows like reduceByKeyAndWindow(..., 30, 10) mean the last 30 seconds computed every 10 seconds, and both numbers must be multiples of the batch interval. Output operations such as saveAsTextFiles and foreachRDD write the results out.

Then the class ends at exactly 5 PM, which he admits surprised him too.

My Checklist

These are all the tips from the class in one place, so I can come back to this list without rewatching six hours:

• Aim for at least as many partitions as cores (1,000 to 10,000 is normal), and coalesce or repartition after a big filter.
• Never collect something big. Save it or sample it first. Use parallelize only for tests.
• Cache the cleaned RDD that several actions reuse, not the base RDD you just read, and unpersist it when you’re done.
• Pick MEMORY_ONLY, then MEMORY_ONLY_SER with Kryo, and only use disk when recomputing is expensive. Use MEMORY_AND_DISK for RDDs right after an expensive shuffle.
• Use Kryo in production and register your classes.
• Executor heaps between 8 GB and ~40 GB, and at most ~75% of the machine for Spark, leaving room for HDFS or Cassandra on the same node.
• Cache right after an expensive shuffle, and pass numPartitions to transformations that shuffle anyway.
• Oversubscribe task slots 2 to 3 times the free cores, then measure.
• Put SPARK_LOCAL_DIRS on SSDs, one mount per disk, no RAID.
• In Standalone, always set spark.cores.max (the total for the app, oversubscribed) and spark.executor.memory, and give executors the memory to match their cores.
• On YARN, use client mode for shells and cluster mode for jobs you walk away from, and run the history server.
• Turn on the external shuffle service whenever you use dynamic allocation.
• Learn which transformations shuffle, read toDebugString, and balance the work between parallel stages.
• Use mapValues (or preservesPartitioning) when the keys don’t change.
• When one task in a stage takes far longer than the median, find the hot keys and salt them. Aggregate twice for aggregations, and copy the small side once per salt for joins (or broadcast it if it fits).
• Use one connection per partition, not per element, when writing to a database.
• Broadcast small lookup tables instead of joining them, and use accumulators for counters.
• In PySpark, prefer DataFrames and SQL so the work stays in the JVM. Try PyPy for pure Python code, and raise spark.python.worker.memory if workers run out.
• In streaming, keep processing time under the batch interval, add receivers to ingest more, and use CMS or G1 instead of the parallel GC.
• Check the Storage tab for skew, the Environment tab for which settings won, and thread dumps for hot code.
• Don’t use Spark to answer web requests. Someone asked about it in the Q&A, and his answer was that Spark is a batch engine and isn’t meant for millisecond queries. Run a batch job that fills a database such as Cassandra, and serve the app from that.

1. Standalone has caught up since. spark.executor.cores now lets one worker run several executors of the same app, so the SPARK_WORKER_INSTANCES trick he shows for big machines isn’t needed, and dynamic allocation works in Standalone with the external shuffle service.↩︎

2. These fractions were replaced by unified memory in Spark 1.6. spark.memory.fraction (0.6 by default) is now one pool that caching and execution borrow from each other, and spark.memory.storageFraction (0.5) is the part cached data is protected in. The old settings are gone in Spark 3, but his advice to figure out whose memory ran out still applies. Tachyon has also been renamed Alluxio, and OFF_HEAP now uses Spark’s own off-heap memory.↩︎

3. That flag exists on PySpark’s map. In Scala, use mapValues, which keeps the partitioner on its own.↩︎

4. Adaptive query execution came in Spark 3.0 and is on by default since 3.2. The skew handling is spark.sql.adaptive.skewJoin.enabled, which is on once AQE is on, and it decides what counts as skewed with skewedPartitionFactor and skewedPartitionThresholdInBytes.↩︎

5. The hash-based shuffle was removed in Spark 2.0, so sort-based is the only one now. The off-heap Unsafe work from this benchmark also grew into Project Tungsten, which is why DataFrames manage their own memory today.↩︎

6. DStreams still work, but Structured Streaming, built on DataFrames, is what Spark recommends for new streaming jobs.↩︎