【hadoop代碼筆記】Mapreduce shuffle過程之Map輸出過程

一、概要描述 shuffle是MapReduce的一個核心過程，因此沒有在前面的MapReduce作業提交的過程中描述，而是單獨拿出來比較詳細的描述。 根據官方的流程圖示如下：

 

 本篇文章中只是想嘗試從代碼分析來說明在map端是如何將map的輸出保存下來等待reduce來取。 在執行每個map task時，無論map方法中執行什么邏輯，最終都是要把輸出寫到磁盤上。如果沒有reduce階段，則直接輸出到hdfs上，如果有有reduce作業，則每個map方法的輸出在寫磁盤前線在內存中緩存。每個map task都有一個環狀的內存緩沖區，存儲着map的輸出結果，默認100m，在每次當緩沖區快滿的時候由一個獨立的線程將緩沖區的數據以一個溢出文件的方式存放到磁盤，當整個map task結束后再對磁盤中這個map task產生的所有溢出文件做合並，被合並成已分區且已排序的輸出文件。然后等待reduce task來拉數據。

二、 流程描述

1. 在child進程調用到runNewMapper時，會設置output為NewOutputCollector，來負責map的輸出。
2. 在map方法的最后，不管經過什么邏輯的map處理，最終一般都要調用到TaskInputOutputContext的write方法，進而調用到設置的output即NewOutputCollector的write方法
3. NewOutputCollector其實只是對MapOutputBuffer的一個封裝，其write方法調用的是MapOutputBuffer的collect方法。
4. MapOutputBuffer的collect方法中把key和value序列化后存儲在一個環形緩存中，如果緩存滿了則會調用startspill方法設置信號量，使得一個獨立的線程SpillThread可以對緩存中的數據進行處理。
5. SpillThread線程的run方法中調用sortAndSpill方法對緩存中的數據進行排序后寫溢出文件。
6. 當map輸出完成后，會調用output的close方法。
7. 在close方法中調用flush方法，對剩余的緩存進行處理，最后調用mergeParts方法，將前面過程的多個溢出文件合並為一個。

 

Mapreduce shuffle過程之Map輸出過程代碼流程

三、代碼詳細

1 MapTask的runNewMapper方法 注意到有這樣一段代碼。即當job中只有map沒有reduce的時候，這個rg.apache.hadoop.mapreduce.RecordWriter類型的對象 output是一Outputformat中定義的writer，即直接寫到輸出中。如果是有Reduce，則output是一個NewOutputCollector類型輸出。

 

if (job.getNumReduceTasks() == 0) {
        output = outputFormat.getRecordWriter(taskContext);
      } else {
        output = new NewOutputCollector(taskContext, job, umbilical, reporter);
      }
      mapperContext = contextConstructor.newInstance(mapper, job, getTaskID(),input, output, committer,                                                 reporter, split);
      input.initialize(split, mapperContext);
      mapper.run(mapperContext);

 

和其他的RecordWriter一樣，NewOutputCollector也繼承自RecordWriter抽象類。除了一個close方法釋放資源外，該抽象類定義的最主要的方法就一個void write(K key, V value)。即寫入key，value。

2. Mapper的run方法，對每個輸出執行map方法。

public void run(Context context) throws IOException, InterruptedException {
    setup(context);
    while (context.nextKeyValue()) {
      map(context.getCurrentKey(), context.getCurrentValue(), context);
    }
    cleanup(context);
}

 

3. Mapper的map方法，默認是直接把key和value寫入

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

一般使用中會做很多我們需要的操作，如著名的wordcount中，把一行單詞切分后，數一（value都設為one = new IntWritable(1)），但最終都是要把結果寫入。即調用context.write(key,value)

public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }

4.TaskInputOutputContext的write方法。調用的是contex中的RecordWriter的write方法。即調用的是NewOutputCollector的write方法。

public void write(KEYOUT key, VALUEOUT value
                    ) throws IOException, InterruptedException {
    output.write(key, value);
}

5.NewOutputCollector的write方法。

public void write(K key, V value) throws IOException, InterruptedException {
      collector.collect(key, value,
                        partitioner.getPartition(key, value, partitions));
}

從方法名上不難看出提供寫數據的是MapOutputCollector<K,V>類型的 collector對象從NewOutputCollector的構造函數中看到collector的初始化。

collector = new MapOutputBuffer<K,V>(umbilical, job, reporter);

6.MapOutputBuffer的構造函數，在了解MapOutputBuffer的collect方法前，先了解下期構造函數，看做了哪些初始化。

public MapOutputBuffer(TaskUmbilicalProtocol umbilical, JobConf job,
                           TaskReporter reporter
                           ) throws IOException, ClassNotFoundException {
      this.job = job;
      this.reporter = reporter;
      localFs = FileSystem.getLocal(job);
     //1）設定map的分區數，即作業 配置中的的reduce數
      partitions = job.getNumReduceTasks();

      rfs = ((LocalFileSystem)localFs).getRaw();

      indexCacheList = new ArrayList<SpillRecord>();

      //2）重要的參數
      final float spillper = job.getFloat("io.sort.spill.percent",(float)0.8);
      final float recper = job.getFloat("io.sort.record.percent",(float)0.05);
      final int sortmb = job.getInt("io.sort.mb", 100);
      if (spillper > (float)1.0 || spillper < (float)0.0) {
        throw new IOException("Invalid \"io.sort.spill.percent\": " + spillper);
      }
      if (recper > (float)1.0 || recper < (float)0.01) {
        throw new IOException("Invalid \"io.sort.record.percent\": " + recper);
      }
      if ((sortmb & 0x7FF) != sortmb) {
        throw new IOException("Invalid \"io.sort.mb\": " + sortmb);
      }
      //3)sorter,使用其對map的輸出在partition內進行內排序。
      sorter = ReflectionUtils.newInstance(
            job.getClass("map.sort.class", QuickSort.class, IndexedSorter.class), job);
      LOG.info("io.sort.mb = " + sortmb);
      // buffers and accounting
     //把單位是M的sortmb設定左移20，還原單位為個
      int maxMemUsage = sortmb << 20;
      int recordCapacity = (int)(maxMemUsage * recper);
      recordCapacity -= recordCapacity % RECSIZE;
      //輸出緩存
      kvbuffer = new byte[maxMemUsage - recordCapacity];
      bufvoid = kvbuffer.length;
      recordCapacity /= RECSIZE;
      kvoffsets = new int[recordCapacity];
      kvindices = new int[recordCapacity * ACCTSIZE];
      softBufferLimit = (int)(kvbuffer.length * spillper);
      softRecordLimit = (int)(kvoffsets.length * spillper);
      LOG.info("data buffer = " + softBufferLimit + "/" + kvbuffer.length);
      LOG.info("record buffer = " + softRecordLimit + "/" + kvoffsets.length);
      // k/v serialization
      comparator = job.getOutputKeyComparator();
      keyClass = (Class<K>)job.getMapOutputKeyClass();
      valClass = (Class<V>)job.getMapOutputValueClass();
      serializationFactory = new SerializationFactory(job);
      keySerializer = serializationFactory.getSerializer(keyClass);
      keySerializer.open(bb);
      valSerializer = serializationFactory.getSerializer(valClass);
      valSerializer.open(bb);
      // counters
      mapOutputByteCounter = reporter.getCounter(MAP_OUTPUT_BYTES);
      mapOutputRecordCounter = reporter.getCounter(MAP_OUTPUT_RECORDS);
      Counters.Counter combineInputCounter = 
        reporter.getCounter(COMBINE_INPUT_RECORDS);
      combineOutputCounter = reporter.getCounter(COMBINE_OUTPUT_RECORDS);
      // 4）compression
      if (job.getCompressMapOutput()) {
        Class<? extends CompressionCodec> codecClass =
          job.getMapOutputCompressorClass(DefaultCodec.class);
        codec = ReflectionUtils.newInstance(codecClass, job);
      }
      // 5）combiner是一個NewCombinerRunner類型，調用Job的reducer來對map的輸出在map端進行combine。
      combinerRunner = CombinerRunner.create(job, getTaskID(), 
                                             combineInputCounter,
                                             reporter, null);
      if (combinerRunner != null) {
        combineCollector= new CombineOutputCollector<K,V>(combineOutputCounter);
      } else {
        combineCollector = null;
      }
      //6)啟動一個SpillThread線程來
      minSpillsForCombine = job.getInt("min.num.spills.for.combine", 3);
      spillThread.setDaemon(true);
      spillThread.setName("SpillThread");
      spillLock.lock();
      try {
        spillThread.start();
        while (!spillThreadRunning) {
          spillDone.await();
        }
      } catch (InterruptedException e) {
        throw (IOException)new IOException("Spill thread failed to initialize"
            ).initCause(sortSpillException);
      } finally {
        spillLock.unlock();
      }
      if (sortSpillException != null) {
        throw (IOException)new IOException("Spill thread failed to initialize"
            ).initCause(sortSpillException);
      }
}

7.MapOutputBuffer的collect方法。

參數partition是partitioner根據key計算得到的當前key value屬於的partition索引。寫key和value寫入緩存，當緩存滿足spill條件時，通過調用startSpill方法設置變量並通過spillReady.signal()，通知spillThread；並等待spill結束（通過spillDone.await()等待）緩沖區的作用是批量收集map結果，減少磁盤IO的影響。key/value對以及Partition的結果都會被寫入緩沖區。寫入之前，key與value值都會被序列化成字節數組。kvindices保持了記錄所屬的分區，key在緩沖區開始的位置和value在緩沖區開始的位置，通過kvindices，可以在緩沖區中找到對應的記錄。

輸出緩沖區中，和kvstart，kvend和kvindex對應的是bufstart，bufend和bufmark。這部分還涉及到變量bufvoid，用於表明實際使用的緩沖區結尾和變量bufmark，用於標記記錄的結尾。需要bufmark，是因為key或value的輸出是變長的。

[caption id="attachment_539" align="alignnone" width="739"] Key Value序列化后緩存[/caption]

public synchronized void collect(K key, V value, int partition
                                     ) throws IOException {
      reporter.progress();
      if (key.getClass() != keyClass) {
        throw new IOException("Type mismatch in key from map: expected "
                              + keyClass.getName() + ", recieved "
                              + key.getClass().getName());
      }
      if (value.getClass() != valClass) {
        throw new IOException("Type mismatch in value from map: expected "
                              + valClass.getName() + ", recieved "
                              + value.getClass().getName());
      }
//對kvoffsets的長度取模，暗示我們這是一個環形緩存。
final int kvnext = (kvindex + 1) % kvoffsets.length;
     //進入臨界區 
spillLock.lock();
      try {
        boolean kvfull;
        do {
          if (sortSpillException != null) {
            throw (IOException)new IOException("Spill failed"
                ).initCause(sortSpillException);
          }
          // sufficient acct space
          kvfull = kvnext == kvstart;
          final boolean kvsoftlimit = ((kvnext > kvend)
              ? kvnext - kvend > softRecordLimit
              : kvend - kvnext <= kvoffsets.length - softRecordLimit);
          if (kvstart == kvend && kvsoftlimit) {
            LOG.info("Spilling map output: record full = " + kvsoftlimit);
//其實是設置變量並通過spillReady.signal()，通知spillThread；並等待spill結束
            startSpill();
          }
          if (kvfull) {
            try {
              while (kvstart != kvend) {
//kvstart不等於kvend，表示系統正在spill,等待spillDone信號
                reporter.progress();
                spillDone.await();
              }
            } catch (InterruptedException e) {
              throw (IOException)new IOException(
                  "Collector interrupted while waiting for the writer"
                  ).initCause(e);
            }
          }
        } while (kvfull);
      } finally {
        spillLock.unlock();
      }

      try {
//先對key串行化，然后對value做串行化，臨時變量keystart，valstart和valend分別記錄了key結果的開始位置，value結果的開始位置和value結果的結束位置。串行化過程中，往緩沖區寫是最終調用了Buffer.write方法
        // serialize key bytes into buffer
        int keystart = bufindex;
        keySerializer.serialize(key);
        if (bufindex < keystart) {
          //如果key串行化后出現bufindex < keystart，那么會調用BlockingBuffer的reset方法。原因是在spill的過程中需要對<key，value>排序，這種情況下，傳遞給RawComparator的必須是連續的二進制緩沖區，通過BlockingBuffer.reset方法   會把bufvoid設置為bufmark，緩沖區開始部分往后挪，然后將原來位於bufmark到bufvoid出的結果，拷到緩沖區開始處，這樣的話，key串行化的結果就連續存放在緩沖區的最開始處。       
        bb.reset();
          keystart = 0;
        }
        // serialize value bytes into buffer
        final int valstart = bufindex;
        valSerializer.serialize(value);
        int valend = bb.markRecord();

        if (partition < 0 || partition >= partitions) {
          throw new IOException("Illegal partition for " + key + " (" +
              partition + ")");
        }

        mapOutputRecordCounter.increment(1);
        mapOutputByteCounter.increment(valend >= keystart
            ? valend - keystart
            : (bufvoid - keystart) + valend);

        // update accounting info
        int ind = kvindex * ACCTSIZE;
        kvoffsets[kvindex] = ind;
        kvindices[ind + PARTITION] = partition;
        kvindices[ind + KEYSTART] = keystart;
        kvindices[ind + VALSTART] = valstart;
        kvindex = kvnext;
      } catch (MapBufferTooSmallException e) {
        LOG.info("Record too large for in-memory buffer: " + e.getMessage());
//如果value的串行化結果太大，不能一次放入緩沖區
        spillSingleRecord(key, value, partition);
        mapOutputRecordCounter.increment(1);
        return;
      }
}

8.MapOutputBuffer.BlockingBuffer的reset()方法.

如果key串行化后出現bufindex < keystart，那么會調用BlockingBuffer的reset方法。原因是在spill的過程中需要對<key，value>排序，這種情況下，傳遞給RawComparator的必須是連續的二進制緩沖區，通過BlockingBuffer.reset方法當發現key的串行化結果出現不連續的情況時，會把bufvoid設置為bufmark，緩沖區開始部分往后挪，然后將原來位於bufmark到bufvoid出的結果，拷到緩沖區開始處，這樣的話，key串行化的結果就連續存放在緩沖區的最開始處。

 

BlockingBuffer.reset方法

bufstart前面的緩沖區如果不能放下整個key串行化的結果，，處理的方式是將bufindex置0，然后調用BlockingBuffer內部的out的write方法直接輸出，這實際調用了Buffer.write方法，會啟動spill過程，最終會成功寫入key串行化的結果。

protected synchronized void reset() throws IOException {

        int headbytelen = bufvoid - bufmark;
        bufvoid = bufmark;
        //當發現key的串行化結果出現不連續的情況時，會把bufvoid設置為bufmark，緩沖區開始部分往后挪，然后將原來位於bufmark到bufvoid出的結果，拷到緩沖區開始處，這樣的話，key串行化的結果就連續存放在緩沖區的最開始處。

        if (bufindex + headbytelen < bufstart) {
          System.arraycopy(kvbuffer, 0, kvbuffer, headbytelen, bufindex);
          System.arraycopy(kvbuffer, bufvoid, kvbuffer, 0, headbytelen);
          bufindex += headbytelen;
        } else {
         //bufstart前面的緩沖區如果不能夠放下整個key串行化的結果，處理的方式是將bufindex置0，然后調用BlockingBuffer內部的out的write方法直接輸出
          byte[] keytmp = new byte[bufindex];
          System.arraycopy(kvbuffer, 0, keytmp, 0, bufindex);
          bufindex = 0;
          out.write(kvbuffer, bufmark, headbytelen);
          out.write(keytmp);
        }
      }
    }

 

9. MapOutputBuffer.Buffer的write方法。在key和value序列列化的時候，被調用寫到緩存中。如果spill線程正在把緩存的數據寫溢出文件，則阻塞。

public synchronized void write(byte b[], int off, int len)
          throws IOException {
        boolean buffull = false;
        boolean wrap = false;
        spillLock.lock();
        try {
          do {//循環，直到有足夠的空間可以寫數據
            if (sortSpillException != null) {
              throw (IOException)new IOException("Spill failed"
                  ).initCause(sortSpillException);
            }

            // sufficient buffer space?
            if (bufstart <= bufend && bufend <= bufindex) {
              buffull = bufindex + len > bufvoid;
              wrap = (bufvoid - bufindex) + bufstart > len;
            } else {
              // bufindex <= bufstart <= bufend
              // bufend <= bufindex <= bufstart
              wrap = false;
              buffull = bufindex + len > bufstart;
            }

            if (kvstart == kvend) {
              // spill thread not running
              if (kvend != kvindex) {
                //如果數組中有記錄(kvend != kvindex)，那么，根據需要（目前輸出空間不足或記錄數達到spill條件）啟動spill過程
                final boolean bufsoftlimit = (bufindex > bufend)
                  ? bufindex - bufend > softBufferLimit
                  : bufend - bufindex < bufvoid - softBufferLimit;
                if (bufsoftlimit || (buffull && !wrap)) {
                  LOG.info("Spilling map output: buffer full= " + bufsoftlimit);
                  startSpill();
                }
              } else if (buffull && !wrap) {
               // 如果空間不夠（buffull && !wrap），但是緩存中沒有記錄，表明這個記錄非常大，內存緩沖區不能容下這么大的數據量，拋MapBufferTooSmallException異常，直接寫文件不用寫緩存
                final int size = ((bufend <= bufindex)
                  ? bufindex - bufend
                  : (bufvoid - bufend) + bufindex) + len;
                bufstart = bufend = bufindex = bufmark = 0;
                kvstart = kvend = kvindex = 0;
                bufvoid = kvbuffer.length;
                throw new MapBufferTooSmallException(size + " bytes");
              }
            }

            if (buffull && !wrap) {
              try {
              //如果空間不足但是spill在運行，等待spillDone
                while (kvstart != kvend) {
                  reporter.progress();
                  spillDone.await();
                }
              } catch (InterruptedException e) {
                  throw (IOException)new IOException(
                      "Buffer interrupted while waiting for the writer"
                      ).initCause(e);
              }
            }
          } while (buffull && !wrap);
        } finally {
          spillLock.unlock();
        }
        //真正把數據寫緩存的地方！如果buffull，則寫數據會不連續，則寫滿剩余緩沖區，然后設置bufindex=0，並從bufindex處接着寫。否則，就是從bufindex處開始寫。
        if (buffull) {
          //緩存剩余長度
          final int gaplen = bufvoid - bufindex;
          //把剩余的寫滿
          System.arraycopy(b, off, kvbuffer, bufindex, gaplen);
          //剩下長度
          len -= gaplen;
          //剩下偏移
          off += gaplen;
         //寫指針移到開頭
          bufindex = 0;
        }
        從指定的開頭寫
        System.arraycopy(b, off, kvbuffer, bufindex, len);
        bufindex += len;
      }
    }

 

buffull和wrap條件說明

如圖，對bufful和wrap條件進行說明： 在上面兩種情況下，即情況1和情況2，

buffull = bufindex + len > bufvoid;
wrap = (bufvoid - bufindex) + bufstart > len;

buffull條件判斷為從下次寫指針的位置bufindex到緩存結束bufvoid的空間是否有足夠的空間容納寫的內容，wrap是圖中白顏色部分的空間（前后空白合在一起）是否比輸入大，如果是，wrap為true； 情況3和情況4中，

wrap = false;
buffull = bufindex + len > bufstart;

buffull判斷bufindex到bufstart的空間是否滿足條件，而wrap肯定是false。 條件（buffull && !wrap）滿足時，目前的空間不夠一次寫。

10.MapOutputBuffer 的spillSingleRecord方法。如果在collect方法中處理緩存失敗，則直接把這條記錄些到spill文件中。對應單條記錄即使設置了combiner也不用。如果記錄非常大，內存緩沖區不能容下這么大的數據量，拋MapBufferTooSmallException異常，直接寫文件不用寫緩存。

private void spillSingleRecord(final K key, final V value,
                                   int partition) throws IOException {
      long size = kvbuffer.length + partitions * APPROX_HEADER_LENGTH;
      FSDataOutputStream out = null;
      try {
        // 創建spill文件
        final SpillRecord spillRec = new SpillRecord(partitions);
        final Path filename = mapOutputFile.getSpillFileForWrite(getTaskID(),
            numSpills, size);
        out = rfs.create(filename);

        IndexRecord rec = new IndexRecord();
        for (int i = 0; i < partitions; ++i) {
          IFile.Writer<K, V> writer = null;
          try {
            long segmentStart = out.getPos();
             writer = new IFile.Writer<K,V>(job, out, keyClass, valClass, codec,
                                            spilledRecordsCounter);

            if (i == partition) {
              final long recordStart = out.getPos();
              writer.append(key, value);
              mapOutputByteCounter.increment(out.getPos() - recordStart);
            }
            writer.close();

            // 把偏移記錄在index中
            rec.startOffset = segmentStart;
            rec.rawLength = writer.getRawLength();
            rec.partLength = writer.getCompressedLength();
            spillRec.putIndex(rec, i);

            writer = null;
          } catch (IOException e) {
            if (null != writer) writer.close();
            throw e;
          }
        }
         //如果index滿了，則把index也寫到index文件中。沒滿則把該條index記錄加入到indexCacheList中，並更新totalIndexCacheMemory。
        if (totalIndexCacheMemory >= INDEX_CACHE_MEMORY_LIMIT) {
          // create spill index file
          Path indexFilename = mapOutputFile.getSpillIndexFileForWrite(
              getTaskID(), numSpills,
              partitions * MAP_OUTPUT_INDEX_RECORD_LENGTH);
          spillRec.writeToFile(indexFilename, job);
        } else {
          indexCacheList.add(spillRec);
          totalIndexCacheMemory +=
            spillRec.size() * MAP_OUTPUT_INDEX_RECORD_LENGTH;
        }
        ++numSpills;
      } finally {
        if (out != null) out.close();
      }
    }

 

11.MapOutputBuffer的startSpill。喚醒等待spillReady的線程。 

private synchronized void startSpill() {
      LOG.info("bufstart = " + bufstart + "; bufend = " + bufmark +
               "; bufvoid = " + bufvoid);
      LOG.info("kvstart = " + kvstart + "; kvend = " + kvindex +
               "; length = " + kvoffsets.length);
      kvend = kvindex;
      bufend = bufmark;
      spillReady.signal();
    }

 

12.SpillThread的run方法。

該Thread會檢查內存中的輸出緩存區，在滿足一定條件的時候將緩沖區中的內容spill到硬盤上。這是一個標准的生產者-消費者模型，MapTask的collect方法是生產者，spillThread是消費者，它們之間同步是通過spillLock（ReentrantLock）和spillLock上的兩個條件變量（spillDone和spillReady）完成的。當kvstart == kvend條件成立時，表示沒有要spill的記錄。

public void run() {
      //臨界區
        spillLock.lock();
        spillThreadRunning = true;
        try {
          while (true) {
            spillDone.signal();
當kvstart == kvend條件成立時，表示沒有要spill的記錄
            while (kvstart == kvend) {
              spillReady.await();
            }
            try {
              spillLock.unlock();
              //執行操作
              sortAndSpill();
            } catch (Exception e) {
              sortSpillException = e;
            } catch (Throwable t) {
              sortSpillException = t;
              String logMsg = "Task " + getTaskID() + " failed : " 
                              + StringUtils.stringifyException(t);
              reportFatalError(getTaskID(), t, logMsg);
            } finally {
              spillLock.lock();
              if (bufend < bufindex && bufindex < bufstart) {
                bufvoid = kvbuffer.length;
              }
              kvstart = kvend;
              bufstart = bufend;
            }
          }
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        } finally {
          spillLock.unlock();
          spillThreadRunning = false;
        }
      }
    }

 

13..MapOutputBuffer的sortAndSpill() 方法 SpillThread線程的run方法中調用sortAndSpill把緩存中的輸出寫到格式為+ "/spill" + spillNumber + ".out"的spill文件中。索引（kvindices）保持在spill{spill號}.out.index中，數據保存在spill{spill號}.out中 創建SpillRecord記錄，輸出文件和IndexRecord記錄，然后，需要在kvoffsets上做排序，排完序后順序訪問kvoffsets，也就是按partition順序訪問記錄。按partition循環處理排完序的數組，如果沒有combiner，則直接輸出記錄，否則，調用combineAndSpill，先做combin然后輸出。循環的最后記錄IndexRecord到SpillRecord。

private void sortAndSpill() throws IOException, ClassNotFoundException,
                                       InterruptedException {
      //approximate the length of the output file to be the length of the
      //buffer + header lengths for the partitions
      long size = (bufend >= bufstart
          ? bufend - bufstart
          : (bufvoid - bufend) + bufstart) +
                  partitions * APPROX_HEADER_LENGTH;
      FSDataOutputStream out = null;
      try {
        // 創建溢出文件
        final SpillRecord spillRec = new SpillRecord(partitions);
        final Path filename = mapOutputFile.getSpillFileForWrite(getTaskID(),
            numSpills, size);
        out = rfs.create(filename);

        final int endPosition = (kvend > kvstart)
          ? kvend
          : kvoffsets.length + kvend;
//使用sorter進行排序, 在內存中進行，參照MapOutputBuffer的compare方法實現的這里的排序也是對序列化的字節做的排序。排序是在kvoffsets上面進行，參照MapOutputBuffer的swap方法實現。
        sorter.sort(MapOutputBuffer.this, kvstart, endPosition, reporter);
        int spindex = kvstart;
        IndexRecord rec = new IndexRecord();
        InMemValBytes value = new InMemValBytes();
        for (int i = 0; i < partitions; ++i) {
          IFile.Writer<K, V> writer = null;
          try {
            long segmentStart = out.getPos();
            writer = new Writer<K, V>(job, out, keyClass, valClass, codec,
                                      spilledRecordsCounter);
            if (combinerRunner == null) {
              // 如果沒有combinner則直接寫鍵值
              DataInputBuffer key = new DataInputBuffer();
              while (spindex < endPosition &&
                  kvindices[kvoffsets[spindex % kvoffsets.length]
                            + PARTITION] == i) {
                final int kvoff = kvoffsets[spindex % kvoffsets.length];
                getVBytesForOffset(kvoff, value);
                key.reset(kvbuffer, kvindices[kvoff + KEYSTART],
                          (kvindices[kvoff + VALSTART] - 
                           kvindices[kvoff + KEYSTART]));
                //鍵值寫到溢出文件
                writer.append(key, value);
                ++spindex;
              }
            } else {
              int spstart = spindex;
              while (spindex < endPosition &&
                  kvindices[kvoffsets[spindex % kvoffsets.length]
                            + PARTITION] == i) {
                ++spindex;
              }
              //如果設置了combiner，則調用了combine方法后的結果寫到IFile中，writer還是先前的writer。減少溢寫到磁盤的數據量。
              if (spstart != spindex) {
                combineCollector.setWriter(writer);
                RawKeyValueIterator kvIter =
                  new MRResultIterator(spstart, spindex);
                combinerRunner.combine(kvIter, combineCollector);
              }
            }

            // close the writer
            writer.close();

            // record offsets
            rec.startOffset = segmentStart;
            rec.rawLength = writer.getRawLength();
            rec.partLength = writer.getCompressedLength();
            spillRec.putIndex(rec, i);

            writer = null;
          } finally {
            if (null != writer) writer.close();
          }
        }

        if (totalIndexCacheMemory >= INDEX_CACHE_MEMORY_LIMIT) {
          // 寫溢出索引文件，格式如+ "/spill" + spillNumber +  ".out.index"
          Path indexFilename = mapOutputFile.getSpillIndexFileForWrite(
              getTaskID(), numSpills,
              partitions * MAP_OUTPUT_INDEX_RECORD_LENGTH);
          spillRec.writeToFile(indexFilename, job);
        } else {
          indexCacheList.add(spillRec);
          totalIndexCacheMemory +=
            spillRec.size() * MAP_OUTPUT_INDEX_RECORD_LENGTH;
        }
        LOG.info("Finished spill " + numSpills);
        ++numSpills;
      } finally {
        if (out != null) out.close();
      }
    }

 

14 MapOutputBuffer的compare方法和swap方法 MapOutputBuffer實現了IndexedSortable接口，從接口命名上就可以猜想到，這個排序不是移動數據，而是移動數據的索引。在這里要排序的其實是kvindices對象，通過移動其記錄在kvoffets上的索引來實現。 如圖，表示了寫磁盤前Sort的效果。kvindices保持了記錄所屬的（Reduce）分區，key在緩沖區開始的位置和value在緩沖區開始的位置，通過kvindices，我們可以在緩沖區中找到對應的記錄。kvoffets用於在緩沖區滿的時候對kvindices的partition進行排序，排完序的結果將輸出到輸出到本地磁盤上，其中索引（kvindices）保持在spill{spill號}.out.index中，數據保存在spill{spill號}.out中。通過觀察MapOutputBuffer的compare知道，先是在partition上排序，然后是在key上排序。 kvindices在kvoffets上排序

public int compare(int i, int j) {
      final int ii = kvoffsets[i % kvoffsets.length];
      final int ij = kvoffsets[j % kvoffsets.length];
      // 先在partition上排序
      if (kvindices[ii + PARTITION] != kvindices[ij + PARTITION]) {
        return kvindices[ii + PARTITION] - kvindices[ij + PARTITION];
      }
      // 然后在可以上排序
      return comparator.compare(kvbuffer,
          kvindices[ii + KEYSTART],
          kvindices[ii + VALSTART] - kvindices[ii + KEYSTART],
          kvbuffer,
          kvindices[ij + KEYSTART],
          kvindices[ij + VALSTART] - kvindices[ij + KEYSTART]);
    }

     public void swap(int i, int j) {
      i %= kvoffsets.length;
      j %= kvoffsets.length;
      //通過交互在kvoffsets上的索引達到排序效果
      int tmp = kvoffsets[i];
      kvoffsets[i] = kvoffsets[j];
      kvoffsets[j] = tmp;
    }

 

15. MapOutputBuffer的flush() 方法

Mapper的結果都已經collect了，需要對緩沖區做一些最后的清理，調用flush方法，合並spill{n}文件產生最后的輸出。先等待可能的spill過程完成，然后判斷緩沖區是否為空，如果不是，則調用sortAndSpill，做最后的spill，然后結束spill線程.

public synchronized void flush() throws IOException, ClassNotFoundException,
                                            InterruptedException {
      LOG.info("Starting flush of map output");
      spillLock.lock();
      try {
        while (kvstart != kvend) {
          reporter.progress();
          spillDone.await();
        }
        if (sortSpillException != null) {
          throw (IOException)new IOException("Spill failed"
              ).initCause(sortSpillException);
        }
        if (kvend != kvindex) {
          kvend = kvindex;
          bufend = bufmark;
          sortAndSpill();
        }
      } catch (InterruptedException e) {
        throw (IOException)new IOException(
            "Buffer interrupted while waiting for the writer"
            ).initCause(e);
      } finally {
        spillLock.unlock();
      }
      assert !spillLock.isHeldByCurrentThread();

      try {
        spillThread.interrupt();
        spillThread.join();
      } catch (InterruptedException e) {
        throw (IOException)new IOException("Spill failed"
            ).initCause(e);
      }
      // release sort buffer before the merge
      kvbuffer = null;
      mergeParts();
    }

 

16.MapTask.MapOutputBuffer的mergeParts()方法.

從不同溢寫文件中讀取出來的，然后再把這些值加起來。因為merge是將多個溢寫文件合並到一個文件，所以可能也有相同的key存在，在這個過程中如果配置設置過Combiner，也會使用Combiner來合並相同的key。?mapreduce讓每個map只輸出一個文件，並且為這個文件提供一個索引文件，以記錄每個reduce對應數據的偏移量。

private void mergeParts() throws IOException, InterruptedException, 
                                     ClassNotFoundException {
      // get the approximate size of the final output/index files
      long finalOutFileSize = 0;
      long finalIndexFileSize = 0;
      final Path[] filename = new Path[numSpills];
      final TaskAttemptID mapId = getTaskID();

      for(int i = 0; i < numSpills; i++) {
        filename[i] = mapOutputFile.getSpillFile(mapId, i);
        finalOutFileSize += rfs.getFileStatus(filename[i]).getLen();
      }
      if (numSpills == 1) { //如果只有一個spill文件，則重命名為輸出的最終文件
        rfs.rename(filename[0],
            new Path(filename[0].getParent(), "file.out"));
        if (indexCacheList.size() == 0) {
          rfs.rename(mapOutputFile.getSpillIndexFile(mapId, 0),
              new Path(filename[0].getParent(),"file.out.index"));
        } else {
          indexCacheList.get(0).writeToFile(
                new Path(filename[0].getParent(),"file.out.index"), job);
        }
        return;
      }

      // read in paged indices
      for (int i = indexCacheList.size(); i < numSpills; ++i) {
        Path indexFileName = mapOutputFile.getSpillIndexFile(mapId, i);
        indexCacheList.add(new SpillRecord(indexFileName, job));
      }

      //make correction in the length to include the sequence file header
      //lengths for each partition
      finalOutFileSize += partitions * APPROX_HEADER_LENGTH;
      finalIndexFileSize = partitions * MAP_OUTPUT_INDEX_RECORD_LENGTH;
      Path finalOutputFile = mapOutputFile.getOutputFileForWrite(mapId,
                             finalOutFileSize);
      Path finalIndexFile = mapOutputFile.getOutputIndexFileForWrite(
                            mapId, finalIndexFileSize);

      //The output stream for the final single output file
      FSDataOutputStream finalOut = rfs.create(finalOutputFile, true, 4096);

      if (numSpills == 0) {
        //如果沒有spill文件，則創建一個 dummy files
        IndexRecord rec = new IndexRecord();
        SpillRecord sr = new SpillRecord(partitions);
        try {
          for (int i = 0; i < partitions; i++) {
            long segmentStart = finalOut.getPos();
            Writer<K, V> writer =
              new Writer<K, V>(job, finalOut, keyClass, valClass, codec, null);
            writer.close();
            rec.startOffset = segmentStart;
            rec.rawLength = writer.getRawLength();
            rec.partLength = writer.getCompressedLength();
            sr.putIndex(rec, i);
          }
          sr.writeToFile(finalIndexFile, job);
        } finally {
          finalOut.close();
        }
        return;
      }
      {
        IndexRecord rec = new IndexRecord();
        final SpillRecord spillRec = new SpillRecord(partitions);
        for (int parts = 0; parts < partitions; parts++) {
          //在循環內對每個分區分別創建segment然后做merge
          List<Segment<K,V>> segmentList =
            new ArrayList<Segment<K, V>>(numSpills);
          for(int i = 0; i < numSpills; i++) {
            IndexRecord indexRecord = indexCacheList.get(i).getIndex(parts);

            Segment<K,V> s =
              new Segment<K,V>(job, rfs, filename[i], indexRecord.startOffset,
                               indexRecord.partLength, codec, true);
            segmentList.add(i, s);

            if (LOG.isDebugEnabled()) {
              LOG.debug("MapId=" + mapId + " Reducer=" + parts +
                  "Spill =" + i + "(" + indexRecord.startOffset + "," +
                  indexRecord.rawLength + ", " + indexRecord.partLength + ")");
            }
          }

          //merge
          @SuppressWarnings("unchecked")
          RawKeyValueIterator kvIter = Merger.merge(job, rfs,
                         keyClass, valClass, codec,
                         segmentList, job.getInt("io.sort.factor", 100),
                         new Path(mapId.toString()),
                         job.getOutputKeyComparator(), reporter,
                         null, spilledRecordsCounter);

          //write merged output to disk
         //執行merge，並且把merge結果寫到"/file.out"的最終輸出中去。
          long segmentStart = finalOut.getPos();
          Writer<K, V> writer =
              new Writer<K, V>(job, finalOut, keyClass, valClass, codec,
                               spilledRecordsCounter);
          if (combinerRunner == null || numSpills < minSpillsForCombine) {
            Merger.writeFile(kvIter, writer, reporter, job);
          } else {
            combineCollector.setWriter(writer);
            combinerRunner.combine(kvIter, combineCollector);
          }

          //close
          writer.close();

          // record offsets
          //把index寫到最終的"/file.out.index"文件中。
          rec.startOffset = segmentStart;
          rec.rawLength = writer.getRawLength();
          rec.partLength = writer.getCompressedLength();
          spillRec.putIndex(rec, parts);
        }
        spillRec.writeToFile(finalIndexFile, job);
        finalOut.close();
        for(int i = 0; i < numSpills; i++) {
          rfs.delete(filename[i],true);
        }
      }
    }

 

合並前后index文件和spill文件的結構圖

 merge最終生成一個spill.out和spill.out.index文件

從前面的分析指導，多個partition的都在一個輸出文件中，但是按照partition排序的。即把maper輸出按照partition分段了。一個partition對應一個reducer，因此一個reducer只要獲取一段即可。

完。

參考：

參考並補充了http://caibinbupt.iteye.com/blog/401374文章中關於內存中索引結構的分析。謝謝。

為了維護文章的版本一致、最新、可追溯，轉載請注明： 轉載自idouba本文鏈接地址: 【hadoop代碼筆記】Mapreduce shuffle過程之Map輸出過程