MapReduce框架Mapper和Reducer类源码分析

一:Mapper类

在Hadoop的mapper类中,有4个主要的函数,分别是:setup,cleanup,map,run。代码如下:

  1. protected void setup(Context context) throws IOException, InterruptedException {
  2. // NOTHING
  3. }

  4. protected void map(KEYIN key, VALUEIN value,
  5.                      Context context) throws IOException, InterruptedException {
  6. context.write((KEYOUT) key, (VALUEOUT) value);
  7. }

  8. protected void cleanup(Context context) throws IOException, InterruptedException {
  9. // NOTHING
  10. }

  11. public void run(Context context) throws IOException, InterruptedException {
  12.     setup(context);
  13.     while (context.nextKeyValue()) {
  14.       map(context.getCurrentKey(), context.getCurrentValue(), context);
  15.     }
  16.     cleanup(context);
  17.   }
  18. }
由上面的代码,我们可以了解到,当调用到map时,通常会先执行一个setup函数,最后会执行一个cleanup函数。而默认情况下,这两个函数的内容都是nothing。因此,当map方法不符合应用要求时,可以试着通过增加setup和cleanup的内容来满足应用的需求。

二:Reducer类

在Hadoop的reducer类中,有3个主要的函数,分别是:setup,clearup,reduce。代码如下:
  1.   /**
  2.    * Called once at the start of the task.
  3.    */
  4.   protected void setup(Context context
  5.                        ) throws IOException, InterruptedException {
  6.     // NOTHING
  7.   }


  1.   /**
  2.    * This method is called once for each key. Most applications will define
  3.    * their reduce class by overriding this method. The default implementation
  4.    * is an identity function.
  5.    */
  6.   @SuppressWarnings("unchecked")
  7.   protected void reduce(KEYIN key, Iterable<VALUEIN> values, Context context
  8.                         ) throws IOException, InterruptedException {
  9.     for(VALUEIN value: values) {
  10.       context.write((KEYOUT) key, (VALUEOUT) value);
  11.     }
  12.   }


  1.   /**
  2.    * Called once at the end of the task.
  3.    */
  4.   protected void cleanup(Context context
  5.                          ) throws IOException, InterruptedException {
  6.     // NOTHING
  7.   }


在用户的应用程序中调用到reducer时,会直接调用reducer里面的run函数,其代码如下:
  1. /*
  2.    * control how the reduce task works.
  3.    */
  4.   @SuppressWarnings("unchecked")
  5.   public void run(Context context) throws IOException, InterruptedException {
  6.     setup(context);
  7.     while (context.nextKey()) {
  8.       reduce(context.getCurrentKey(), context.getValues(), context);
  9.       // If a back up store is used, reset it
  10.       ((ReduceContext.ValueIterator)
  11.           (context.getValues().iterator())).resetBackupStore();
  12.     }
  13.     cleanup(context);
  14.   }
  15. }


由上面的代码,我们可以了解到,当调用到reduce时,通常会先执行一个setup函数,最后会执行一个cleanup函数。而默认情况下,这两个函数的内容都是nothing。因此,当reduce不符合应用要求时,可以试着通过增加setup和cleanup的内容来满足应用的需求。

上一篇:JAVA设计模式—适配器模式,装饰模式,代理模式异同


下一篇:PLSQL中隐式打开cursor的小例子