title: Word count sorting in MapReduce Python using yarn comparator
body: <p>I want to solve the word count problem and want to get the results in reverse sorted order according to the frequency of occurrence in the file.</p>
  
  <p>Following are the four files (2 mappers and 2 reducers, as one Map Reduce job cannot solve this problem) I wrote for this purpose:</p>
  
  <p>1) mapper1.py</p>
  
  <pre><code>import sys
  import re
  
  reload(sys)
  sys.setdefaultencoding('utf-8') # required to convert to unicode
  
  for line in sys.stdin:
      try:
          article_id, text = unicode(line.strip()).split('\t', 1)
      except ValueError as e:
          continue
      words = re.split("\W*\s+\W*", text, flags=re.UNICODE)
      for word in words:
          print "%s\t%d" % (word.lower(), 1)
  </code></pre>
  
  <p>2) reducer1.py</p>
  
  <pre><code>import sys
  
  current_key = None
  word_sum = 0
  
  for line in sys.stdin:
      try:
          key, count = line.strip().split('\t', 1)
          count = int(count)
      except ValueError as e:
          continue
      if current_key != key:
          if current_key:
              print "%s\t%d" % (current_key, word_sum)
          word_sum = 0
          current_key = key
      word_sum += count
  
  if current_key:
      print "%s\t%d" % (current_key, word_sum)
  </code></pre>
  
  <p>3) mapper2.py</p>
  
  <pre><code>import sys
  import re
  
  reload(sys)
  sys.setdefaultencoding('utf-8') # required to convert to unicode
  
  
  for line in sys.stdin:
      try:
          word, count = line.strip().split('\t', 1)
          count = int(count)
      except ValueError as e:
          continue
  
      print "%s\t%d" % (word, count)
  </code></pre>
  
  <p>4) reducer2.py</p>
  
  <pre><code>import sys
  
  for line in sys.stdin:
      try:
          word, count = line.strip().split('\t', 1)
          count = int(count)
      except ValueError as e:
          continue
  
      print "%s\t%d" % (word, count)
  </code></pre>
  
  <p>The following are the two yarn commands run by me in a bash environment</p>
  
  <pre><code>OUT_DIR="wordcount_result_1"
  NUM_REDUCERS=8
  
  hdfs dfs -rm -r -skipTrash ${OUT_DIR} &gt; /dev/null
  
  yarn jar /opt/cloudera/parcels/CDH/lib/hadoop-mapreduce/hadoop-streaming.jar \
      -D mapred.jab.name="Streaming wordCount" \
      -D mapreduce.job.reduces=${NUM_REDUCERS} \
      -files mapper1.py,reducer1.py \
      -mapper "python mapper1.py" \
      -combiner "python reducer1.py" \
      -reducer "python reducer1.py" \
      -input /test/articles-part-short \
      -output ${OUT_DIR} &gt; /dev/null
  
  
  OUT_DIR_2="wordcount_result_2"
  NUM_REDUCERS=1
  
  hdfs dfs -rm -r -skipTrash ${OUT_DIR_2} &gt; /dev/null
  
  yarn jar /opt/cloudera/parcels/CDH/lib/hadoop-mapreduce/hadoop-streaming.jar \
      -D mapred.jab.name="Streaming wordCount Rating" \
      -D mapreduce.job.output.key.comparator.class=org.apache.hadoop.mapreduce.lib.partition.KeyFieldBasedComparator \
      -D map.output.key.field.separator=\t \
      -D mapreduce.partition.keycomparator.options=-k2,2nr \
      -D mapreduce.job.reduces=${NUM_REDUCERS} \
      -files mapper2.py,reducer2.py \
      -mapper "python mapper2.py" \
      -reducer "python reducer2.py" \
      -input ${OUT_DIR} \
      -output ${OUT_DIR_2} &gt; /dev/null
  
  hdfs dfs -cat ${OUT_DIR_2}/part-00000 | head
  </code></pre>
  
  <p>This is not giving me the right answer. Can someone please explain where did it go wrong?</p>
  
  <p>On the other hand,</p>
  
  <p>in the <code>mapper2.py</code> if I print in the following manner,</p>
  
  <p><code>print "%d\t%s" % (count, word)</code></p>
  
  <p>and in the <code>reducer2.py</code> if I read in the following manner,</p>
  
  <p><code>count, word = line.strip().split('\t', 1)</code></p>
  
  <p>and edit the 2nd yarn command option to</p>
  
  <p><code>-D mapreduce.partition.keycomparator.options=-k1,1nr</code></p>
  
  <p>it gives me right answer.</p>
  
  <p>Why is it behaving differently in both of the above cases?</p>
  
  <p>Can someone please help me understand the Comparator options of Hadoop MapReduce?</p>