7 Tips for Improving MapReduce Performance

 

Since MapReduce and HDFS are complex distributed systems that run arbitrary user code, there’s no hard and fast set of rules to achieve optimal performance; instead, I tend to think of tuning a cluster or job much like a doctor would treat a sick human being. There are a number of key symptoms to look for, and each set of symptoms leads to a different diagnosis and course of treatment.

 

In medicine, there’s no automatic process that can replace the experience of a well seasoned doctor. The same is true with complex distributed systems — experienced users and operators often develop a “sixth sense” for common issues. Having worked with Cloudera customers in a number of different industries, each with a different workload, dataset, and cluster hardware, I’ve accumulated a bit of this experience, and would like to share some with you today.

 

In this blog post, I’ll highlight a few tips for improving MapReduce performance. The first few tips are cluster-wide, and will be useful for operators and developers alike. The latter tips are for developers writing custom MapReduce jobs in Java. For each tip, I’ll also note a few of the “symptoms” or “diagnostic tests” that indicate a particular remedy might bring you some good improvements.

 

Please note, also, that these tips contain lots of rules of thumb based on my experience across a variety of situations. They may not apply to your particular workload, dataset, or cluster, and you should always benchmark your jobs before and after any changes. For these tips, I’ll show some comparative numbers for a 40GB wordcount job on a small 4-node cluster. Tuned optimally, each of the map tasks in this job runs in about 33 seconds, and the total job runtime is about 8m30s.

 

 

Tip 1) Configure your cluster correctly

Diagnostics/symptoms:

• top shows slave nodes fairly idle even when all map and reduce task slots are filled up running jobs.
• top shows kernel processes like RAID (mdX_raid*) or pdflush taking most of the CPU time.
• Linux load averages are often seen more than twice the number of CPUs on the system.
• Linux load averages stay less than half the number of CPUs on the system, even when running jobs.
• Any swap usage on nodes beyond a few MB.

The first step to optimizing your MapReduce performance is to make sure your cluster configuration has been tuned. For starters, check out our earlier blog post on configuration parameters. In addition to those knobs in the Hadoop configuration, here are a few more checklist items you should go through before beginning to tune the performance of an individual job:

• Make sure the mounts you’re using for DFS and MapReduce storage have been mounted with the noatime option. This disables access time tracking and can improve IO performance.
• Avoid RAID and LVM on TaskTracker and DataNode machines – it generally reduces performance.
• Make sure you’ve configured mapred.local.dir and dfs.data.dir to point to one directory on each of your disks to ensure that all of your IO capacity is used. Run iostat -dx 5 from the sysstat package while the cluster is loaded to make sure each disk shows utilization.
• Ensure that you have SMART monitoring for the health status of your disk drives. MapReduce jobs are fault tolerant, but dying disks can cause performance to degrade as tasks must be re-executed. If you find that a particular TaskTracker becomes blacklisted on many job invocations, it may have a failing drive.
• Monitor and graph swap usage and network usage with software like Ganglia. Monitoring Hadoop metrics in Ganglia is also a good idea. If you see swap being used, reduce the amount of RAM allocated to each task in mapred.child.java.opts.

Benchmarks:

Unfortunately I was not able to perform benchmarks for this tip, as it would involve re-imaging the cluster. If you have had relevant experience, feel free to leave a note in the Comments section below.

Tip 2) Use LZO Compression

Diagnostics/symptoms:

• This is almost always a good idea for intermediate data! In the doctor analogy, consider LZO compression your vitamins.
• Output data size of MapReduce job is nontrivial.
• Slave nodes show high iowait utilization in top and iostat when jobs are running.

Almost every Hadoop job that generates an non-negligible amount of map output will benefit from intermediate data compression with LZO. Although LZO adds a little bit of CPU overhead, the reduced amount of disk IO during the shuffle will usually save time overall.

Whenever a job needs to output a significant amount of data, LZO compression can also increase performance on the output side. Since writes are replicated 3x by default, each GB of output data you save will save 3GB of disk writes.
In order to enable LZO compression, check out our recent guest blog from Twitter. Be sure to set mapred.compress.map.output to true.

Benchmarks:

Disabling LZO compression on the wordcount example increased the job runtime only slightly on our cluster. The FILE_BYTES_WRITTEN counter increased from 3.5GB to 9.2GB, showing that the compression yielded a 62% decrease in disk IO. Since this job was not sharing the cluster, and each node has a high ratio of number of disks to number of tasks, IO is not the bottleneck here, and thus the improvement was not substantial. On clusters where disks are pegged due to a lot of concurrent activity, a 60% reduction in IO can yield a substantial improvement in job completion speed.

Tip 3) Tune the number of map and reduce tasks appropriately

Diagnostics/symptoms:

• Each map or reduce task finishes in less than 30-40 seconds.
• A large job does not utilize all available slots in the cluster.
• After most mappers or reducers are scheduled, one or two remains pending and then runs all alone.

Tuning the number of map and reduce tasks for a job is important and easy to overlook. Here are some rules of thumb I use to set these parameters:

• If each task takes less than 30-40 seconds, reduce the number of tasks. The task setup and scheduling overhead is a few seconds, so if tasks finish very quickly, you’re wasting time while not doing work. JVM reuse can also be enabled to solve this problem.
• If a job has more than 1TB of input, consider increasing the block size of the input dataset to 256M or even 512M so that the number of tasks will be smaller. You can change the block size of existing files with a command like hadoop distcp -Ddfs.block.size=$[256*1024*1024] /path/to/inputdata /path/to/inputdata-with-largeblocks. After this command completes, you can remove the original data.
• So long as each task runs for at least 30-40 seconds, increase the number of mapper tasks to some multiple of the number of mapper slots in the cluster. If you have 100 map slots in your cluster, try to avoid having a job with 101 mappers – the first 100 will finish at the same time, and then the 101st will have to run alone before the reducers can run. This is more important on small clusters and small jobs.
• Don’t schedule too many reduce tasks – for most jobs, we recommend a number of reduce tasks equal to or a bit less than the number of reduce slots in the cluster.

  Benchmarks:

  To make the wordcount job run with too many tasks, I ran it with the argument -Dmapred.max.split.size=$[16*1024*1024]. This yielded 2640 tasks instead of the 360 that the framework chose by default. When running with this setting, each task took about 9 seconds, and watching the Cluster Summary view on the JobTracker showed the number of running maps fluctuating between 0 and 24 continuously throughout the job. The entire job finished in 17m52s, more than twice as slow as the original job.

  Tip 4) Write a Combiner

  Diagnostics/symptoms:

• A job performs aggregation of some sort, and the Reduce input groups counter is significantly smaller than the Reduce input records counter.
• The job performs a large shuffle (e.g. map output bytes is multiple GB per node)
• The number of spilled records is many times larger than the number of map output records as seen in the Job counters.

If your algorithm involves computing aggregates of any sort, chances are you can use a Combiner in order to perform some kind of initial aggregation before the data hits the reducer. The MapReduce framework runs combiners intelligently in order to reduce the amount of data that has to be written to disk and transfered over the network in between the Map and Reduce stages of computation.

Benchmarks:

I modified the word count example to remove the call to setCombinerClass, and otherwise left it the same. This changed the average map task run time from 33s to 48s, and increased the amount of shuffled data from 1GB to 1.4GB. The total job runtime increased from 8m30s to 15m42s, nearly a factor of two. Note that this benchmark was run with map output compression enabled – without map output compression, the effect of the combiner would have been even more important.

Tip 5) Use the most appropriate and compact Writable type for your data

Symptoms/diagnostics:

• Text objects are used for working with non-textual or complex data
• IntWritable or LongWritable objects are used when most output values tend to be significantly smaller than the maximum value.

When users are new to programming in MapReduce, or are switching from Hadoop Streaming to Java MapReduce, they often use the Text writable type unnecessarily. Although Text can be convenient, converting numeric data to and from UTF8 strings is inefficient and can actually make up a significant portion of CPU time. Whenever dealing with non-textual data, consider using the binary Writables like IntWritable, FloatWritable, etc.

In addition to avoiding the text parsing overhead, the binary Writable types will take up less space as intermediate data. Since disk IO and network transfer will become a bottleneck in large jobs, reducing the sheer number of bytes taken up by the intermediate data can provide a substantial performance gain. When dealing with integers, it can also sometimes be faster to use VIntWritable or VLongWritable — these implement variable-length integer encoding which saves space when serializing small integers. For example, the value 4 will be serialized in a single byte, whereas the value 10000 will be serialized in two. These variable length numbers can be very effective for data like counts, where you expect that the majority of records will have a small number that fits in one or two bytes.

If the Writable types that ship with Hadoop don’t fit the bill, consider writing your own. It’s pretty simple, and will be significantly faster than parsing text. If you do so, make sure to provide a RawComparator — see the source code for the built in Writables for an example.

Along the same vein, if your MapReduce job is part of a multistage workflow, use a binary format like SequenceFile for the intermediate steps, even if the last stage needs to output text. This will reduce the amount of data that needs to be materialized along the way.

Benchmarks:

For the example word count job, I modified the intermediate count values to be Text type rather than IntWritable. In the reducer, I used Integer.parseString(value.toString()) when accumulating the sum. The performance of the suboptimal version of the WordCount was about 10% slower than the original. The full job ran in a bit over 9 minutes, and each map task took 36 seconds instead of the original 33. Since integer parsing is itself rather fast, this did not represent a large improvement; in the general case, I have seen using more efficient Writables to make as much as a 2-3x difference in performance.

 

Tip 6) Reuse Writables

Symptoms/diagnostics:

• Add -verbose:gc -XX:+PrintGCDetails to mapred.child.java.opts. Then inspect the logs for some tasks. If garbage collection is frequent and represents a lot of time, you may be allocating unnecessary objects.
• grep for “new Text” or “new IntWritable” in your code base. If you find this in an inner loop, or inside the map or reduce functions this tip may help.
• This tip is especially helpful when your tasks are constrained in RAM.

One of the first mistakes that many MapReduce users make is to allocate a new Writable object for every output from a mapper or reducer. For example, one might implement a word-count mapper like this:

public void map(...) {
  ...
  for (String word : words) {
    output.collect(new Text(word), new IntWritable(1));
  }
}

This implementation causes thousands of very short-lived objects to be allocated. While the Java garbage collector does a reasonable job at dealing with this, it is more efficient to write:

class MyMapper ... {
  Text wordText = new Text();
  IntWritable one = new IntWritable(1);
  public void map(...) {
    ...
    for (String word : words) {
      wordText.set(word);
      output.collect(word, one);
    }
  }
}

Benchmarks:

When I modified the word count example as described above, I initially found it made no difference in the run time of the job. This is because this cluster’s default settings include a 1GB heap size for each task, so garbage collection never ran. However, running it with each task allocated only 200mb of heap size showed a drastic slowdown in the version that did not reuse Writables — the total job runtime increased from around 8m30s to over 17 minutes. The original version, which does reuse Writables, stayed the same speed even with the smaller heap. Since reusing Writables is an easy fix, I recommend always doing so – it may not bring you a gain for every job, but if you’re low on memory it can make a huge difference.

Tip 7) Use “Poor Man’s Profiling” to see what your tasks are doing

This is a trick I almost always use when first looking at the performance of a MapReduce job. Profiling purists will disagree and say that this won’t work, but you can’t argue with results!

In order to do what I call “poor man’s profiling”, ssh into one of your slave nodes while some tasks from a slow job are running. Then simply run sudo killall -QUIT java 5-10 times in a row, each a few seconds apart. Don’t worry — this doesn’t cause anything to quit, despite the name. Then, use the JobTracker interface to navigate to the stdout logs for one of the tasks that’s running on this node, or look in /var/log/hadoop/userlogs/ for a stdout file of a task that is currently running. You’ll see stack trace output from each time you sent the SIGQUIT signal to the JVM.

It takes a bit of experience to parse this output, but here’s the method I usually use:

1. For each thread in the trace, quickly scan for the name of your Java package (e.g. com.mycompany.mrjobs). If you don’t see any lines in the trace that are part of your code, skip over this thread.
2. When you find a stack trace that has some of your code in it, make a quick mental note what it’s doing. For example, “something NumberFormat-related” is all you need at this point. Don’t worry about specific line numbers yet.
3. Go down to the next dump you took a few seconds later in the logs. Perform the same process here and make a note.
4. After you’ve gone through 4-5 of the traces, you might notice that the same vague thing shows up in every one of them. If that thing is something that you expect to be fast, you probably found your culprit. If you take 10 traces, and 5 of them show NumberFormat in the dump, it means that you’re spending somewhere around 50% of your CPU time formatting numbers, and you might consider doing something differently.

Sure, this method isn’t as scientific as using a real profiler on your tasks, but I’ve found that it’s a surefire way to notice any glaring CPU bottlenecks very quickly and with no setup involved. It’s also a technique that you’ll get better at with practice as you learn what a normal dump looks like and when something jumps out as odd.

Here are a few performance mistakes I often find through this technique:

• NumberFormat is slow – avoid it where possible.
• String.split, as well as encoding or decoding UTF8 are slower than you think – see above tips about using the appropriate Writables
• Concatenating Strings rather than using StringBuffer.append

 

References

http://blog.cloudera.com/blog/2009/12/7-tips-for-improving-mapreduce-performance/