Tuning Spark: partitioning, shuffling and caching

1. Partitioning: the two levers

Chapter 6 introduced the RDD as a resilient distributed dataset, and glossed over the word “distributed”. This chapter pays that word its due: the lab deck on Spark optimization starts where any tuning conversation has to start, with partitioning.

Data in RDDs are split into multiple partitions. One property holds unconditionally and explains most of what follows:

Given that constraint, partitioning serves two purposes, and the lab deck is careful to attach a separate control knob to each:

• Parallelize computation across workers and CPU cores — controlled by the number of partitions.
• Minimize network traffic for data exchange between executors — controlled by the partitioning criteria (the partitioner).

The two knobs answer two different questions. How many pieces is the data cut into? decides how much parallelism you get. Which piece does each key go to? decides how much data has to travel between executors when operations that group by key run.

2. The number of partitions

If you do not specify it, Spark sets the number of partitions automatically. The lab deck gives the exact rules, and they are a mini-review of everything Chapter 6 said about where RDDs come from:

• In-memory datasets (e.g., parallelize) → total number of available cores.
• External datasets (e.g., files read from HDFS) → one partition for each block of each file — Chapter 3’s block size, reused as the natural data unit.
• Transformations → largest number of partitions in a parent RDD.

Why does the number matter so much? Because of a one-line equation the deck states plainly: one task is run for each partition. Partition count is task count. Everything that follows is the consequence of that mapping.

Too few partitions

• Less concurrency: some cores will not be used — with four partitions and eight cores, half the cluster idles.
• Risks from data skewness: the job might get stuck on that one task handling most of the keys — the straggler problem of Chapter 6, now caused by partitioning rather than by scheduling.

Too many partitions

• Excessive overhead in managing/scheduling many small tasks — each task is a scheduling decision, a JVM object, a status report; thousands of two-millisecond tasks pay that overhead on every single one.

The deck then gives the rules of thumb, and they are worth memorising as a ladder:

3. The partitioning criteria

The second lever is the partitioning criteria: the rule that decides, for a key/value RDD, which key goes to which partition. The deck gives it three jobs:

• Put values that belong to the same key in the same partition — this is what makes local aggregation possible.
• Define how the key should be distributed in different partitions — ideally evenly, so no partition becomes a straggler.
• There is deliberately no control over the specific worker node a partition goes to — placement is the cluster’s business, not yours.

Where does the criteria come from? It depends on how the key/value RDD is built:

• In-memory datasets → none (no specific criteria);
• External datasets → none (no specific criteria);
• Transformations → depends on the transformation (possibly none).

And the crucial pricing note: forcing a partitioning criteria on an RDD requires shuffling the data — it costs. A criteria is not a free annotation; it is a redistribution, and redistribution is the expensive operation that sections 5 and 6 will dissect.

There are three ways of partitioning:

4. Repartitioning: pay once, shuffle less

Given that a criteria costs a shuffle, when is it worth paying? The deck gives two answers:

• In presence of an unbalanced dataset — the skew of section 2, fixed by re-cutting the pieces;
• When the dataset is reused multiple times in key-oriented operations (e.g., reduceByKey, join) — because partitioning avoids re-shuffling the data every time.

The second case gets a full worked example, and it is the canonical tuning story of this whole chapter. An application keeps in memory a large table of user information, where UserInfo contains the list of topics the user is subscribed to:

val sc = new SparkContext(...)
val userData = sc.sequenceFile[UserID, UserInfo]("hdfs://...").persist()

Every 5 minutes, userData is combined with a smaller file containing the websites visited by the users; LinkInfo is a single website:

def processNewLogs(logFileName: String) {
  val events = sc.sequenceFile[UserID, LinkInfo](logFileName)
  val joined = userData.join(events) // RDD of (UserID, (UserInfo, LinkInfo)) pairs
  val offTopicVisits = joined.filter {
    case (userId, (userInfo, linkInfo)) => // Expand the tuple into its components
      !userInfo.topics.contains(linkInfo.topic)
  }.count()
  println("Number of visits to non-subscribed topics: " + offTopicVisits)
}

The fix is a one-liner, and it is the whole lesson of the chapter in miniature: partition userData before persisting it.

After the fix, only the events RDD is shuffled, sending events with a certain UserID to the machine that contains the corresponding hash partition of userData. Fewer network communications, and — in the deck’s words — a significant boost in performance.

Repartitioning advantages, in full

The deck then generalises, and the generalisation is a ladder of three cases:

• Operations on a single, partitioned RDD (e.g., reduceByKey) → the values for each key are computed locally (no further shuffling occurs).
• Operations on two RDDs (e.g., join) → if one RDD is partitioned, only the other is shuffled.
• If both RDDs share the same partitioner → partitions of RDD2 are only sent to the correct partition of RDD1 — no full redistribution of either side.
• If both RDDs share the same partitioner and the partitions reside in the same machine (e.g., RDD2 was created with RDD1.mapValues) → no shuffling occurs.

Partitioning heritage: Spark knows what it created

The last piece of the partitioning model is bookkeeping: Spark knows internally how each of its operations affects partitioning, and records it on the result RDD. The partitioner is automatically set on RDDs created by operations that partition the data:

cogroup(), groupWith(), join(), leftOuterJoin(), rightOuterJoin(),
groupByKey(), reduceByKey(), combineByKey(), partitionBy(), sort()

And if the parent RDD has a partitioner, also these narrow transformations keep it:

mapValues(), flatMapValues(), filter()

Everything else — notably map() — produces a result with no partitioner. The reason is semantic: map() can change the key arbitrarily, so the old criteria would be a lie. The heritage rules are exactly the deck’s way of saying the partitioner survives only where the key survives unchanged.

5. Shuffling: mechanism and cost

Everything so far has been circling one word. Time to name it: shuffling is the mechanism used to re-distribute data across partitions. The deck’s definition comes with two qualifiers that set the tone of the whole chapter:

• It is necessary to compute some operations — you cannot join or aggregate by key without bringing the keys together;
• It is complex and costly — and the complexity is about to be itemised.

First, why it is necessary at all. Values for a key live scattered across partitions; all partitions must be read in order to find values for the key k1:

Which operations trigger all this? The deck’s taxonomy is short and examinable:

• Repartitioning (e.g., repartition, coalesce) — always a shuffle by construction.
• ByKey operations (e.g., groupByKey, reduceByKey) — if data is already partitioned by key, no shuffling occurs; and countByKey does not shuffle at all, it just sends counts to the driver.
• Joins (e.g., cogroup, join) — the partitioning of the two RDDs may decrease the amount of shuffled data (section 4’s ladder, in one line).

6. Three generations of shuffle

Shuffling is costly enough that Spark has spent its whole history making it cheaper. The deck tells the story as three generations: hash shuffle (the very first technique), sort shuffle (default with Spark 1.2), and Tungsten sort (an evolution of the sort shuffle).

Hash shuffle: one file per (mapper, reducer)

Each map task creates a file for every reducer. The deck’s example is famous for a reason — a real Yahoo workload with 46,000 mappers and 46,000 reducers:

An evolved version pooled the files per executor: each executor holds a pool of files, one group containing a file for every reducer, with as many groups as the mappers that can run in parallel in the executor itself. With 100 executors, 10 cores each, 1 core per task:

• each executor holds 10 groups of 46,000 files;
• multiplied by the number of executors: over 46,000,000 intermediate files.

Better than two billion, still absurd.

Pros: fast — no sorting is required at all, no hash table maintained; no memory/CPU overhead for sorting the data. Cons: when the amount of partitions is big, performance starts to degrade due to the big amount of output files — a big amount of files written to the filesystem causes IO skew towards random IO, which is in general up to 100× slower than sequential IO.

Sort shuffle: one spill file per mapper

The second generation inverts the file structure. Each mapper keeps output in memory, spills to disk if necessary; each mapper spills to its own file(s); each file is sorted by reducer. When a reducer asks for its data, the “pieces” from each file are collected, sorted in memory, and sent to the reducer.

Same Yahoo example: ~46,000 intermediate files. Not two billion, not forty-six million — one per mapper.

Pros: smaller amount of files created on the map side; smaller amount of random IO operations, mostly sequential writes and reads. Cons: sorting is slower than hashing — although in practice sort shuffle is usually better; and the deck notes the one environment where the trade flips: in case SSD drives are used for the temporary data of Spark shuffles, hash shuffle might work better.

Tungsten sort: shuffle on serialized bytes

The third generation attacks serialization, the third cost of Plate 8.3. Tungsten sort directly works on serialized records: records can be sorted, merged, concatenated, spilled to disk without de-serializing and re-serializing, using a cache-efficient sorter. The pros are one line: it improves performances of the sort shuffle technique. The cons are the interesting part — it can be adopted only under specific conditions:

• No map-side aggregation (i.e., combining) is requested;
• The serializer supports relocation (i.e., reordering) of serialized values (e.g., KryoSerializer);
• Other implementative limits: the shuffle produces fewer than 16,777,216 output partitions, and no individual record is larger than 128 MB when serialized.

7. Caching: persist and storage levels

The third tuning target is memory, and the starting observation is blunt: Spark recomputes an RDD each time an action is called on it. The deck’s example is the “trivial” one that everyone hits:

rddCapraKvLength.count()     // action 1: the whole lineage runs
rddCapraKvLength.collect()    // action 2: the whole lineage runs again

Two actions, two full recomputations of the same RDD. Especially expensive for iterative algorithms — the ones Chapter 9 will build on, where the same dataset is re-read on every iteration.

The fix is persist(): when you persist an RDD x, each node stores in memory the partitions of x that it computes and reuses them in other actions on x. Three properties complete the picture:

• This allows future actions to be much faster;
• Unlike actions, persisting operations do not trigger the computation of the RDD — persistence is a directive on the lineage, not a job;
• To release persisted RDDs, call the unpersist() method.

RDDs can be persisted using the persist() or cache() methods; persist() allows specifying the storage level; cache() uses MEMORY_ONLY. The storage levels form a ladder from pure memory to pure disk:

The deck closes the topic with explicit guidance on which storage level is best:

• It is best to stay in-memory as much as possible;
• Serialization makes the objects much more space-efficient;
• Spill to disk only if the dataset is computed via expensive functions — disk is cheaper than recomputation, but only then;
• Use replication only if you want fast fault recovery.

8. Shared variables: broadcast and accumulators

Variables used inside closures have a problem that Chapter 6 never mentioned: variables used within functions passed to distributed operations (e.g., map or reduce) are copied to every task. The deck adds the design rationale in one clause: supporting general read/write shared variables across tasks would be inefficient. Keeping every task’s copy coherent with the driver — or with each other — is exactly the distributed-systems nightmare the chapters on consensus would warn you about.

Spark therefore supports two (limited) types of shared variables, each with its own semantics:

Broadcast variables

Through broadcast variables it is possible to send read-only copies of variables to every executor. The mechanism matters because Spark already ships the required data to each task: any variable your closure captures travels with every task. Broadcasting is therefore useful exactly when that default is wasteful:

• The same variable is used across multiple stages — without broadcasting it is shipped once per stage; with broadcasting it is shipped only once;
• A large object is used by every task — it is shipped only to the executor and kept in memory for every task to read, instead of being copied into every task’s payload.

val dictionary = Map(("man"-> "noun"), ("is"->"verb"), ("mortal"->"adjective"))
val broadDictionary = sc.broadcast(dictionary)

val result = words
  .map( word => getElementsCount(word, broadDictionary) )
  .reduceByKey(_+_)

This is also the mechanism behind Chapter 7’s Broadcast Hash join: the small table is shipped to every executor once — the same reasoning, at the engine level.

Accumulators

Accumulators are shared variables that can be updated from inside tasks. From the point of view of tasks they are write-only — reading the current value from inside a task would require keeping all tasks up to date, which is the inefficiency the design avoids. They are useful for debugging, computing metrics or commutative and associative functions — i.e., updates whose order does not matter, exactly like the combiners of Chapter 4.

val emptyLines = sc.accumulator(0)

val tokens = textFile.flatMap( line => {
    if (line == "")
       emptyLines += 1
    line.split(" ")
  }
)

Two properties matter for the exam, and the second one is a genuine trap:

• Accumulators do not change the lazy evaluation model of Spark — they are updated only when an action actually runs the transformation;
• Pitfall: if a transformation gets executed multiple times (e.g., because of failures or stragglers), then accumulators are updated multiple times — a re-executed task increments the counter again, and the final value is silently wrong.

Finally, a visibility detail: accumulators will be displayed in the web UI for the stage that modifies that accumulator — the UI is where you actually see the double-counting happening.

9. Tuning the cluster: CPU

The lab deck on Spark optimization ends with caching; a companion deck on cluster configuration turns the same knobs at the level of the whole application. This section and the next summarise that deck, because partitioning, shuffling and caching all happen inside an executor budget you get to choose.

Two main resources: CPU and memory. Disk and network I/O play an important part too, but with a decisive caveat: neither Spark nor YARN currently do anything to actively manage them — everything the chapters so far called “IO cost” is yours to tune indirectly. Every executor in an application has the same fixed:

• Number of cores — controls the number of concurrent tasks an executor can run;
• Heap size — impacts the amount of data that can be cached and the maximum sizes of the shuffle data structures.

The number of executors is also fixed — unless dynamic allocation is enabled.

Tuning CPU is done by setting two options of spark-submit: --num-executors (how many executors) and --executor-cores (how many cores each). The deck adds four things to bear in mind:

• The HDFS client has trouble with tons of concurrent threads — Cloudera’s recommendation is to not exceed 5 tasks per executor;
• Single-core executors throw away the benefits that come from running multiple tasks in a single JVM (e.g., sharing broadcast variables among tasks);
• The application master takes up its own resources that must be budgeted in — in yarn-client mode it defaults to 1 core (and 1GB of RAM); in yarn-cluster mode the application master runs the driver, so you should bolster the driver’s resources with the --driver-memory and --driver-cores properties;
• Each node should be left with enough power to run the OS and Hadoop daemons.

These constraints collapse into a small recipe, and the recipe is examinable arithmetic:

--executor-cores = 3 to 5
                  (such that all cores in a node (-1 for daemons) are employed)

executors-per-node = ⌊ ( available cores - 1 ) / --executor-cores ⌋
--num-executors    = number-of-nodes × executors-per-node - 1
                     (leave out 1 executor for the application master)

10. Tuning the cluster: memory

Tuning memory is done with one main option: --executor-memory, the amount of memory per executor. The cautions are the mirror of the CPU ones:

• Executors with too much memory often result in excessive garbage collection delays — 64GB is a rough guess at a good upper limit for a single executor;
• Each node should be left with enough memory to run the OS and Hadoop daemons;
• The full memory request for each executor is determined by heap size + off heap size — the requested amount is not just the heap.

The off-heap default is small but not negligible: max(384 MB, 10% of spark.executor.memory). The recipe, with off-heap size = 0.1 (10%):

--executor-memory = ( available memory × 0.75 ) × ( 1 - 0.1 ) / executors-per-node
                    (leave out 25% of RAM for running daemons)

And the mapping table that ties every knob to its spark-submit option — this is the one to memorise because exams love pairing a property with its flag:

Test your knowledge