title: How does cleanup() method work?
body: <p>I am currently new to Hadoop. So I have this solved piece of code in MapReduce which finds out the "parts of a country with most 'Data Engineer' jobs for each year" (for example, if the data of the format <strong><em>(Year,Region,Count(Jobs))</em></strong> is <strong><em>"2016,'XYZ',35"</em></strong> and <strong><em>"2016,'ABC',25"</em></strong> and <strong><em>"2015,'sdf',14"</em></strong>, the answer would be <strong>"2016,'XYZ',35"</strong> and <strong>"2015,'sdf',14"</strong>), but I am unable to understand the part in the reducer which is as follows:-</p>
  
  <pre><code>    if (Top5DataEngineer.size() &gt; 1)
              Top5DataEngineer.remove(Top5DataEngineer.firstKey());
      }//Ignore this bracket for the time being.
  
      protected void cleanup(Context context) throws IOException,
              InterruptedException {
          for (Text t : Top5DataEngineer.descendingMap().values())
              context.write(NullWritable.get(), t);
      }
  </code></pre>
  
  <p>This is the full code:-</p>
  
  <pre><code>    import java.io.IOException;
      import org.apache.hadoop.io.LongWritable;
      import org.apache.hadoop.io.Text;
      import org.apache.hadoop.mapreduce.Mapper;
      import org.apache.hadoop.io.NullWritable;
      import org.apache.hadoop.mapreduce.Job;
      import org.apache.hadoop.conf.Configuration;
      import org.apache.hadoop.fs.Path;
      import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
      import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
      import org.apache.hadoop.mapreduce.Partitioner;
      import java.util.TreeMap;
      import org.apache.hadoop.mapreduce.Reducer;
  
      public class Q_002a {
       public static class Q_002a_Mapper extends
          Mapper&lt;LongWritable, Text, Text, LongWritable&gt; {
      LongWritable one = new LongWritable(1);
  
      public void map(LongWritable key, Text values, Context context)
              throws IOException, InterruptedException {
          try {
              if (key.get() &gt; 0)
  
              {
  
                  String[] token = values.toString().split("\t");
                  if (token[4].equals("DATA ENGINEER")) {
                      Text answer = new Text(token[8] + "\t" + token[7]);
                      context.write(answer, one);
                  }
              }
          } catch (ArrayIndexOutOfBoundsException e) {
              System.out.println(e.getMessage());
          } catch (ArithmeticException e1) {
              System.out.println(e1.getMessage());
  
          }
  
      }
  
  }
  
  public static class Q_002a_Partitioner extends Partitioner&lt;Text, LongWritable&gt; {
      @Override
      public int getPartition(Text key, LongWritable value, int numReduceTasks) {
          String[] str = key.toString().split("\t");
          if (str[1].equals("2011"))
              return 0;
          if (str[1].equals("2012"))
              return 1;
          if (str[1].equals("2013"))
              return 2;
          if (str[1].equals("2014"))
              return 3;
          if (str[1].equals("2015"))
              return 4;
          if (str[1].equals("2016"))
              return 5;
          else
              return 6;
      }
  }
  
  public static class Q_002a_Reducer extends
          Reducer&lt;Text, LongWritable, NullWritable, Text&gt; {
      private TreeMap&lt;LongWritable, Text&gt; Top5DataEngineer = new TreeMap&lt;LongWritable, Text&gt;();
      long sum = 0;
  
      public void reduce(Text key, Iterable&lt;LongWritable&gt; values,
              Context context) throws IOException, InterruptedException {
          sum = 0;
          for (LongWritable val : values) {
              sum += val.get();
          }
          Top5DataEngineer.put(new LongWritable(sum), new Text(key + ","
                  + sum));
          if (Top5DataEngineer.size() &gt; 1)
              Top5DataEngineer.remove(Top5DataEngineer.firstKey());
      }
  
      protected void cleanup(Context context) throws IOException,
              InterruptedException {
          for (Text t : Top5DataEngineer.descendingMap().values())
              context.write(NullWritable.get(), t);
      }
  }
  
  public static void main(String args[]) throws IOException,
          InterruptedException, ClassNotFoundException {
      Configuration conf = new Configuration();
      Job job = Job.getInstance(conf, "Top  5 Data Engineer in a worksite");
  
      job.setJarByClass(Q_002a.class);
      job.setMapperClass(Q_002a_Mapper.class);
      job.setPartitionerClass(Q_002a_Partitioner.class);
      job.setReducerClass(Q_002a_Reducer.class);
  
      job.setNumReduceTasks(6);
  
      job.setMapOutputKeyClass(Text.class);
      job.setMapOutputValueClass(LongWritable.class);
  
      job.setOutputKeyClass(NullWritable.class);
      job.setOutputValueClass(Text.class);
  
      FileInputFormat.addInputPath(job, new Path(args[0]));
      FileOutputFormat.setOutputPath(job, new Path(args[1]));
      System.exit(job.waitForCompletion(true) ? 0 : 1);
  
  }
  }
  </code></pre>
  
  <p>This is the output I am getting:-</p>
  
  <p><a href="https://i.stack.imgur.com/pdYdO.jpg" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/pdYdO.jpg" alt="enter image description here"></a></p>
  
  <p>EDIT:- I tried running the code inside the cleanup() method in the reduce() method, but it was not working as expected. It only ran fine when it was in the cleanup() method. Any help regarding this would be appreciated. </p>