title: Hadoop - Reducers spending a lot of time writing data (multiple outputs)
body: <p>So I am using <code>MultipleOutputs</code> from the package <code>org.apache.hadoop.mapreduce.lib.output</code>.</p>
  
  <p>I have a reducer that is doing a join of 2 data sources and emitting 3 different outputs.</p>
  
  <p>55 reduce tasks were invoked and on an average each of them took about 6 minutes to emit data. There were outliers that took about 11 minutes.</p>
  
  <p>So I observed that if I comment the pieces where actual output is happening, i.e. the call to <code>mos.write()</code> (multiple output) then the average time reduces to seconds and the whole job completes in about 2 minutes.</p>
  
  <p>I do have a lot of data to emit (Approximately 40-50 GBs) of data.</p>
  
  <p>What can I do to speed up things a bit, with and without considering compression.</p>
  
  <p>Details: I am using <code>TextOutputFormat</code> and giving a hdfs path/uri.</p>
  
  <p>Further clarifications:</p>
  
  <p>I have small input data to my reducer, however the reducers are doing a reduce side join and hence emitting large amount of data. Since an outlier reducer is approximately taking about 11 minutes, reducing the number of reducers will increase this time and hence increase the overall time of my job and won't solve my purpose. </p>
  
  <p>Input to the reducer comes from 2 mappers.
  Mapper 1 -> Emits about 10,000 records. (Key Id)
  Mapper 2 -> Emits about 15M records. (Key Id, Key Id2, Key Id3)</p>
  
  <p>In reducer I get everything belonging to Key Id, sorted by Key Id, KeyId2 and KeyId3.</p>
  
  <p>So I know I have an iterator which is like:
  Mapper1 output and then Mapper2 output.</p>
  
  <p>Here I store Mapper1 output in an <code>ArrayList</code> and start streaming Mapper2's output.</p>
  
  <p>For every Mapper2 record I do, a <code>mos.write(....)</code>
  I conditionally store a part of this record in memory (in a <code>HashSet</code>)
  For every time KeyId2 changes, I do an extra <code>mos.write(...)</code></p>
  
  <p>At the close method of my reducer, I emit if I stored anything in the conditional step. So third <code>mos.write(...)</code></p>
  
  <p>I have gone through the article <a href="http://blog.cloudera.com/blog/2009/12/7-tips-for-improving-mapreduce-performance/" rel="nofollow">http://blog.cloudera.com/blog/2009/12/7-tips-for-improving-mapreduce-performance/</a> 
  as mentioned:</p>
  
  <p>Tip1 : Configuring my cluster correctly is beyond my control.</p>
  
  <p>Tip2 : Use LZO compression - or compression in general. Something I am trying alongside.</p>
  
  <p>Tip3 : Tune the number of mappers and reducers - My mappers finish really fast (in order of seconds) Probably because they are almost identity mappers. Reducers take some time, as mentioned above (This is the time I'm trying to reduce) So increasing the number of reducers will probably help me - but then there will be resource contention and some reducers will have to wait. This is more of an experimental try and error sort of stuff for me.</p>
  
  <p>Tip4 : Write a combiner. Does not apply to my case (of reduce side joins)</p>
  
  <p>Tip5 : Use apt writable - I need to use Text as of now. All these 3 outputs are going into directories that have a hive schema sitting on top of it. Later when I figure how to emit ParquetFormat files from multiple outputs, I might change this and the tables storage method.</p>
  
  <p>Tip6 : Reuse writables. Okay this is something I have not considered so far, but I still believe that its the disk writes that are taking time and not processing or java heap. But anyway I'll give it a shot again.</p>
  
  <p>Tip7 : Use poor man's profiling. Kind of already done that and figured out that its actually mos.write steps that are taking most of the time.</p>