MapReduce原理分析之MapTask读取数据

ゝ一世哀愁。 2022-02-23 02:34 271阅读 0赞

通过前面的内容介绍相信大家对于MapReduce的操作有了一定的了解，通过客户端源码的分析也清楚了split是逻辑分区，记录了每个分区对应的是哪个文件，从什么位置开始到什么位置介绍，而且一个split对应一个Map Task任务，而MapTask具体是怎么读取文件的呢？本文来具体分析下。

# MapTask读取数据的过程 #

我们要分析的就是如下的过程:

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]

## 1.自定义Mapper ##

在自定义的Mapper中我们只需要重写map方法，那么每读取一行记录就会调用一次该方法，如下

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 1]

## 2.查看Mapper源码 ##

public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> { 
    
      public abstract class Context
        implements MapContext<KEYIN,VALUEIN,KEYOUT,VALUEOUT> { 
      }
      protected void setup(Context context
                           ) throws IOException, InterruptedException { 
        // NOTHING
      }
      @SuppressWarnings("unchecked")
      protected void map(KEYIN key, VALUEIN value, 
                         Context context) throws IOException, InterruptedException { 
        context.write((KEYOUT) key, (VALUEOUT) value);
      }
    
      protected void cleanup(Context context
                             ) throws IOException, InterruptedException { 
        // NOTHING
      }
      
      public void run(Context context) throws IOException, InterruptedException { 
        setup(context);
        try { 
          while (context.nextKeyValue()) { 
            map(context.getCurrentKey(), context.getCurrentValue(), context);
          }
        } finally { 
          cleanup(context);
        }
      }
    }

通过源码我们能够看到里面的方法如下

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>setup</td> 
   <td align="left">任务开始的时候执行一次，可以做一些预处理操作</td> 
  </tr> 
  <tr> 
   <td>map</td> 
   <td align="left">每读取一行记录执行一次，获取KV对</td> 
  </tr> 
  <tr> 
   <td>cleanup</td> 
   <td align="left">任务结束前执行一次，可以做一些收尾工作，比如setup的时候声明一个变量sum，map阶段累加，最后在cleanup的时候将sum输出，获取统计的结果</td> 
  </tr> 
  <tr> 
   <td>run</td> 
   <td align="left">当于map task的驱动，将前面的方法组成了一个模板</td> 
  </tr> 
 </tbody> 
</table>

## 3.context.nextKeyValue() ##

context的类型是Context实现了MapContext

public abstract class Context
        implements MapContext<KEYIN,VALUEIN,KEYOUT,VALUEOUT> { 
      }

MapContext的实现类是MapContextImpl

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 2]

而nextKeyValue()方法的调用

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 3]  
说明调用的是RecordReader中的方法，而具体是RecordReader中的哪个实现类呢？继续往下。

## 4.FileInputFormat ##

我们在启动类中设置了输入输出路径。进入FileInputFormat的子类TextFileInputFormat中查看

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 4]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 5]

说明nextKeyValue()其实执行的是RecordReader中的nextKeyValue方法。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 6]

读取split文件中每行数据的方法。将每行的偏移量保存在key中，每行的具体数据保存在value中，分别通过getCurrentKey方法和getCurrentValue方法来获取。

public boolean nextKeyValue() throws IOException { 
        if (key == null) { 
          key = new LongWritable();
        }
        key.set(pos);
        if (value == null) { 
          value = new Text();
        }
        int newSize = 0;
        // We always read one extra line, which lies outside the upper
        // split limit i.e. (end - 1)
        while (getFilePosition() <= end || in.needAdditionalRecordAfterSplit()) { 
          if (pos == 0) { 
            newSize = skipUtfByteOrderMark();
          } else { 
            newSize = in.readLine(value, maxLineLength, maxBytesToConsume(pos));
            pos += newSize;
          }
    
          if ((newSize == 0) || (newSize < maxLineLength)) { 
            break;
          }
    
          // line too long. try again
          LOG.info("Skipped line of size " + newSize + " at pos " + 
                   (pos - newSize));
        }
        if (newSize == 0) { 
          key = null;
          value = null;
          return false;
        } else { 
          return true;
        }
      }

自此我们简单的分析了下map具体是怎么一行一行的读取数据的了。MapReduce的其他阶段我们后续再慢慢分析

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]: /images/20220223/4209682d60c3477b82fd4f2502ad0da9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 1]: /images/20220223/03d9060798d9487aa86ba1a92098f9f9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 2]: /images/20220223/564c973a43664f87a74edd35eb0b1ef2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 3]: /images/20220223/ed1383c2d67c40878937a89e4406de10.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 4]: /images/20220223/ad9e67051de948db927cbc211b490df7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 5]: /images/20220223/4cc6737e3fe94152b20fcdaa9767d5db.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 6]: /images/20220223/2d0d889f9bd94f65bf86be2ae9ff7ddd.png