大數(shù)據(jù)：Map終結(jié)和Spill文件合并

當(dāng)Mapper沒有數(shù)據(jù)輸入，mapper.run中的while循環(huán)會調(diào)用context.nextKeyValue就返回false，于是便返回到runNewMapper中，在這里程序會關(guān)閉輸入通道和輸出通道，這里關(guān)閉輸出通道并沒有關(guān)閉collector，必須要先flush一下。

獲取更多大數(shù)據(jù)視頻資料請加QQ群：947967114 代碼結(jié)構(gòu)：

Maptask.runNewMapper->NewOutputCollector.close->MapOutputBuffer.flush

我們看flush幫我們做了什么事情，為什么要flush。

public void flush() throws IOException, ClassNotFoundException,

       InterruptedException {

  LOG.info("Starting flush of map output");

  spillLock.lock();

  try {

    while (spillInProgress) {

      reporter.progress();

      spillDone.await();

//這里查看spillInProgress狀態(tài)，如果有spill就等待完成，并且報告狀態(tài)。

    }

    checkSpillException();

    final int kvbend = 4 * kvend;

//kvend是元數(shù)據(jù)塊的終點(diǎn)，元數(shù)據(jù)是向下伸展的。

//kvend是以整數(shù)計的數(shù)組下標(biāo)，kvbend是以字節(jié)計的數(shù)組下標(biāo)

    if ((kvbend + METASIZE) % kvbuffer.length !=

        equator - (equator % METASIZE)) {

//這個條件說明緩沖區(qū)中原來有數(shù)據(jù)，現(xiàn)在spill已經(jīng)完成，需要釋放空間。 獲取更多大數(shù)據(jù)視頻資料請加QQ群：947967114

      // spill finished

//spill一次需要調(diào)整一些參數(shù)，以釋放空間，這個工作通過resetSpill完成

      resetSpill();

private void resetSpill() {

  final int e = equator;

  bufstart = bufend = e;

  final int aligned = e - (e % METASIZE);

  // set start/end to point to first meta record

  // Cast one of the operands to long to avoid integer overflow

  kvstart = kvend = (int)

    (((long)aligned - METASIZE + kvbuffer.length) % kvbuffer.length) / 4;

  LOG.info("(RESET) equator " + e + " kv " + kvstart + "(" +

    (kvstart * 4) + ")" + " kvi " + kvindex + "(" + (kvindex * 4) + ")");

}

//這里其實(shí)就是在調(diào)整各個參數(shù)的位置。比如原點(diǎn)位，kvstart等。

    }

    if (kvindex != kvend) {

//再來判斷緩沖區(qū)是否為空，如果不空表示不滿足spill條件（80%），但map處理完成沒有數(shù)據(jù)輸入。

      kvend = (kvindex + NMETA) % kvmeta.capacity();

      bufend = bufmark;

      LOG.info("Spilling map output");

      LOG.info("bufstart = " + bufstart + "; bufend = " + bufmark +

               "; bufvoid = " + bufvoid);

      LOG.info("kvstart = " + kvstart + "(" + (kvstart * 4) +

               "); kvend = " + kvend + "(" + (kvend * 4) +

               "); length = " + (distanceTo(kvend, kvstart,

                     kvmeta.capacity()) + 1) + "/" + maxRec);

      sortAndSpill();

//調(diào)用一次sortAndSpill過程。 獲取更多大數(shù)據(jù)視頻資料請加QQ群：947967114

    }

  } catch (InterruptedException e) {

    throw new IOException("Interrupted while waiting for the writer", e);

  } finally {

    spillLock.unlock();

  }

//至此所有數(shù)據(jù)都已經(jīng)溢寫出去，緩沖區(qū)已空，所有數(shù)據(jù)都spill到文件中

  assert !spillLock.isHeldByCurrentThread();

  // shut down spill thread and wait for it to exit. Since the preceding

  // ensures that it is finished with its work (and sortAndSpill did not

  // throw), we elect to use an interrupt instead of setting a flag.

  // Spilling simultaneously from this thread while the spill thread

  // finishes its work might be both a useful way to extend this and also

  // sufficient motivation for the latter approach.

  try {

    spillThread.interrupt();

//讓spill線程不在運(yùn)行

    spillThread.join();

//結(jié)束spill線程

  } catch (InterruptedException e) {

    throw new IOException("Spill failed", e);

  }

  // release sort buffer before the merge

  kvbuffer = null;

  mergeParts();

//合并spill文件

  Path outputPath = mapOutputFile.getOutputFile();

  fileOutputByteCounter.increment(rfs.getFileStatus(outputPath).getLen());

}

flush的目的，首先讓緩沖區(qū)的所有KV對數(shù)據(jù)都進(jìn)入spill文件，因?yàn)槊看蝧pill都會產(chǎn)生一個spill文件，所有spill文件可能不止一個，所以要把spill文件合并到單個文件中，分發(fā)給reduce。

所以如果有spill正在進(jìn)行必須等待其完成，也可能沒有spill但是緩沖區(qū)非空，需要再一次sortAndSpill，總之要把緩沖區(qū)清空為止。所有數(shù)據(jù)都spill完成后就可以進(jìn)行mergeParts了

代碼結(jié)構(gòu)：

Maptask.runNewMapper--->NewOutputCollector.close--->MapOutputBuffer.flush--->MapOutputBuffer.mergeParts

源代碼如下：

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];

//每次溢寫都會有一個文件，所以數(shù)組的大小是numSpills。 獲取更多大數(shù)據(jù)視頻資料請加QQ群：947967114

  final TaskAttemptID mapId = getTaskID();

  for(int i = 0; i < numSpills; i++) {

//統(tǒng)計所有這些文件合并之后的大小

    filename[i] = mapOutputFile.getSpillFile(i);

//通過spill文件的編號獲取到指定的spill文件路徑

    finalOutFileSize += rfs.getFileStatus(filename[i]).getLen();//獲取文件大小

  }

  if (numSpills == 1) {

//合并輸出有倆文件一個是output/file.out，一個是output/file.out.index

    sameVolRename(filename[0],

mapOutputFile.getOutputFileForWriteInVolume(filename[0]));

//換個文件名，在原文件名上加個file.out

    if (indexCacheList.size() == 0) {

//索引塊緩存indexCacheList已空

      sameVolRename(mapOutputFile.getSpillIndexFile(0),            mapOutputFile.getOutputIndexFileForWriteInVolume(filename[0]));//spillIndexFile改名。

    } else {

//索引塊緩存indexCacheList中還有索引記錄，要寫到索引文件

      indexCacheList.get(0).writeToFile(

//寫入文件

mapOutputFile.getOutputIndexFileForWriteInVolume(filename[0]), job);

    }

    sortPhase.complete();

    return;

//如果只有一個spill合并已經(jīng)完成。 獲取更多大數(shù)據(jù)視頻資料請加QQ群：947967114

  }

  // read in paged indices

  for (int i = indexCacheList.size(); i < numSpills; ++i) {

//如果spill文件不止一個，需要合并

    Path indexFileName = mapOutputFile.getSpillIndexFile(i);

    indexCacheList.add(new SpillRecord(indexFileName, job));

//先把所有的SpillIndexFile收集在一起。

  }

  //make correction in the length to include the sequence file header

  //lengths for each partition

  finalOutFileSize += partitions * APPROX_HEADER_LENGTH;

//每個partition都有header

  finalIndexFileSize = partitions * MAP_OUTPUT_INDEX_RECORD_LENGTH;

//IndexFile，每個partition一個記錄。

  Path finalOutputFile =

     mapOutputFile.getOutputFileForWrite(finalOutFileSize);

  Path finalIndexFile =

      mapOutputFile.getOutputIndexFileForWrite(finalIndexFileSize);

  //The output stream for the final single output file

  FSDataOutputStream finalOut = rfs.create(finalOutputFile, true, 4096);

//創(chuàng)建合并，最終輸出。

  if (numSpills == 0) {

//要是沒有SipillFile生成，也創(chuàng)建一個空文件

    //create dummy files

    IndexRecord rec = new IndexRecord();

//創(chuàng)建索引記錄

    SpillRecord sr = new SpillRecord(partitions);

//創(chuàng)建spill記錄

    try {

      for (int i = 0; i < partitions; i++) {

        long segmentStart = finalOut.getPos();

        FSDataOutputStream finalPartitionOut = CryptoUtils.wrapIfNecessary(job, finalOut);

        Writer<K, V> writer =

          new Writer<K, V>(job, finalPartitionOut, keyClass, valClass, codec, null);

        writer.close();

//創(chuàng)建后馬上關(guān)閉，形成空文件。

        rec.startOffset = segmentStart;

        rec.rawLength = writer.getRawLength() + CryptoUtils.cryptoPadding(job);

        rec.partLength = writer.getCompressedLength() + CryptoUtils.cryptoPadding(job);

        sr.putIndex(rec, i);

      }

      sr.writeToFile(finalIndexFile, job);

//所以記錄寫入索引文件

    } finally {

      finalOut.close();

    }

    sortPhase.complete();

    return;

  }

  {

    sortPhase.addPhases(partitions); // Divide sort phase into sub-phases

    IndexRecord rec = new IndexRecord();

    final SpillRecord spillRec = new SpillRecord(partitions);

    for (int parts = 0; parts < partitions; parts++) {

//finalOut最終輸出文件。循環(huán)分區(qū)獲得所有spill文件的該分區(qū)數(shù)據(jù)，合并寫入finalOut

      //create the segments to be merged

      List<Segment<K,V>> segmentList =

        new ArrayList<Segment<K, V>>(numSpills);

//創(chuàng)建Segment，數(shù)據(jù)段

      for(int i = 0; i < numSpills; i++) {

//準(zhǔn)備合并所有的Spill文件

        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);

//把每個Spill文件中相同partition的區(qū)段位置收集起來。 獲取更多大數(shù)據(jù)視頻資料請加QQ群：947967114

        if (LOG.isDebugEnabled()) {

          LOG.debug("MapId=" + mapId + " Reducer=" + parts +

              "Spill =" + i + "(" + indexRecord.startOffset + "," +

              indexRecord.rawLength + ", " + indexRecord.partLength + ")");

        }

      }

      int mergeFactor = job.getInt(JobContext.IO_SORT_FACTOR, 100);

//做merge操作時同時操作的stream數(shù)上限

      boolean sortSegments = segmentList.size() > mergeFactor;

      //對segment進(jìn)行排序

      @SuppressWarnings("unchecked")

      RawKeyValueIterator kvIter = Merger.merge(job, rfs,

                     keyClass, valClass, codec,

                     segmentList, mergeFactor,

                     new Path(mapId.toString()),

                     job.getOutputKeyComparator(), reporter, sortSegments,

                     null, spilledRecordsCounter, sortPhase.phase(),

                     TaskType.MAP);

//合并同一partition在所有spill文件中的內(nèi)容，可能還需要sort，合并后的結(jié)構(gòu)是一個序列。

      //write merged output to disk

      long segmentStart = finalOut.getPos();

      FSDataOutputStream finalPartitionOut = CryptoUtils.wrapIfNecessary(job, finalOut);

      Writer<K, V> writer =

          new Writer<K, V>(job, finalPartitionOut, keyClass, valClass, codec,

                           spilledRecordsCounter);

      if (combinerRunner == null || numSpills < minSpillsForCombine) { // minSpillsForCombine在MapOutputBuffer構(gòu)造函數(shù)內(nèi)被初始化，numSpills 為mapTask已經(jīng)溢寫到磁盤spill文件數(shù)量

        Merger.writeFile(kvIter, writer, reporter, job);

//將合并后的結(jié)果直接寫入文件。下面看一下writeFile的源代碼;

public static <K extends Object, V extends Object>

void writeFile(RawKeyValueIterator records, Writer<K, V> writer,

             Progressable progressable, Configuration conf)

throws IOException {

long progressBar = conf.getLong(JobContext.RECORDS_BEFORE_PROGRESS,

    10000);

long recordCtr = 0;

while(records.next()) {

  writer.append(records.getKey(), records.getValue());

//追加的方式輸出到writer中

  if (((recordCtr++) % progressBar) == 0) {

    progressable.progress();

  }

}

回到主代碼：

      } else {

//有combiner

        combineCollector.setWriter(writer);

//就插入combiner環(huán)節(jié)

        combinerRunner.combine(kvIter, combineCollector);

//將合并的結(jié)果經(jīng)過combiner后寫入文件

      }

      //close

      writer.close();//關(guān)閉writer通道

      sortPhase.startNextPhase();

      // record offsets

      rec.startOffset = segmentStart;

//從當(dāng)前段的起點(diǎn)開始

      rec.rawLength = writer.getRawLength() + CryptoUtils.cryptoPadding(job);

      rec.partLength = writer.getCompressedLength() + CryptoUtils.cryptoPadding(job);

      spillRec.putIndex(rec, parts);

    }

    spillRec.writeToFile(finalIndexFile, job);

//把spillFile寫入合并的indexFle

    finalOut.close();

//關(guān)閉最終輸出流

    for(int i = 0; i < numSpills; i++) {

      rfs.delete(filename[i],true);

//刪除所有spill文件

    }

  }

}

該方法會將所有臨時文件合并成一個大文件保存到output/file.out中，同時生成相應(yīng)的索引文件output/file.out.index。 在進(jìn)行文件合并的過程中，Map Task以分區(qū)為單位進(jìn)行合并。對于某個分區(qū)，它將采用多輪遞歸合并的方式：每輪合并io.sort.factor，默認(rèn)是100，個文件，并將產(chǎn)生的文 件重新加入待合并列表中，對文件排序后，重復(fù)上述過程，直到只有一個文件。只生產(chǎn)一個文件可以避免同時打開大量的文件和同時讀取大量的小文件產(chǎn)生的隨機(jī)讀 取帶來的開銷。最后會刪除所有的spill文件。

　　另外需要注意的是，mergeParts()中也有combiner的操作，但是需要滿足一定的條件：1、用戶設(shè)置了combiner；2、spill文件的數(shù)量超過了minSpillsForCombine的值，對應(yīng)配置項"min.num.spills.for.combine"，可自行設(shè)置，默認(rèn)是3。這倆必須同時具備才會在此啟動combiner的本地聚集操作。所以在Map階段有可能combiner會執(zhí)行兩次，所以有可能你的combiner執(zhí)行兩次之后輸出數(shù)據(jù)不符合預(yù)期了。

　　這樣Map階段的任務(wù)就算完成了。主要是讀取數(shù)據(jù)然后寫入內(nèi)存緩沖區(qū)，緩存區(qū)滿足條件就會快排后并設(shè)置partition后，spill到本地文件和索引文件；如果有combiner，spill之前也會做一次聚集操作，待數(shù)據(jù)跑完會通過歸并合并所有spill文件和索引文件，如果有combiner，合并之前在滿足條件后會做一次綜合的聚集操作。map階段的結(jié)果都會存儲在本地中(如果有reducer的話)，非HDFS。

Mapper完成對所有輸入文件的處理，并將緩沖區(qū)的數(shù)據(jù)寫出到spill文件之后，spill文件的存在只有三種可能：沒有spill，一個spill，多個spill。針對這三種都需要一個最終的輸出文件，不管內(nèi)容有沒有，內(nèi)容多少。這個最終文件是和單個spill文件是一樣的，按照partition分成若干段，然后是排好序的KV數(shù)據(jù)，這個merge操作結(jié)合之前的spill文件進(jìn)行sort。就構(gòu)成了一次mergeSort，這個mergeSort只針對同一個Mapper的多個spill文件，以后在Reducer那里還會有Merge針對不同的Mapper文件。

當(dāng)Maptask完成后，從runNewMapper返回，下一個操作就是done。也就是MapTask的收尾工作。MapTask的收尾涉及到怎么把生成的數(shù)據(jù)輸出交給ReduceTask。MapTask和ReduceTask都是擴(kuò)展自Task。但是他們都沒有自己定義done函數(shù)，所以他們都調(diào)用了Task的done。

程序在這里跳出runNewMapper 獲取更多大數(shù)據(jù)視頻資料請加QQ群：947967114

if (useNewApi) {

  runNewMapper(job, splitMetaInfo, umbilical, reporter);

} else {

  runOldMapper(job, splitMetaInfo, umbilical, reporter);

}

done(umbilical, reporter);

這個done我們點(diǎn)進(jìn)去后發(fā)現(xiàn)是Task.done，源碼如下;

public void done(TaskUmbilicalProtocol umbilical,

               TaskReporter reporter

               ) throws IOException, InterruptedException {

LOG.info("Task:" + taskId + " is done."

         + " And is in the process of committing");

updateCounters();

//更新容器

boolean commitRequired = isCommitRequired();

if (commitRequired) {

  int retries = MAX_RETRIES;

  setState(TaskStatus.State.COMMIT_PENDING);

  // say the task tracker that task is commit pending

  while (true) {

    try {

      umbilical.commitPending(taskId, taskStatus);

      break;

//如果commitPending沒有發(fā)生異常，就退出，否則重試。

    } catch (InterruptedException ie) {

      // ignore

    } catch (IOException ie) {

      LOG.warn("Failure sending commit pending: " +

                StringUtils.stringifyException(ie));

      if (--retries == 0) {

        System.exit(67);

      }

    }

  }

  //wait for commit approval and commit

  commit(umbilical, reporter, committer);

}

taskDone.set(true);

reporter.stopCommunicationThread();

// Make sure we send at least one set of counter increments. It's

// ok to call updateCounters() in this thread after comm thread stopped.

updateCounters();

sendLastUpdate(umbilical);

//signal the tasktracker that we are done

sendDone(umbilical);

實(shí)現(xiàn)sendDone的源代碼：

private void sendDone(TaskUmbilicalProtocol umbilical) throws IOException {

int retries = MAX_RETRIES;

while (true) {

  try {

    umbilical.done(getTaskID());

//實(shí)際上這里向MRAppMaster上的TaskAttemptImpl發(fā)送TA_DONE事件

    LOG.info("Task '" + taskId + "' done.");

    return;

  } catch (IOException ie) {

    LOG.warn("Failure signalling completion: " +

             StringUtils.stringifyException(ie));

    if (--retries == 0) {

      throw ie;

    }

  }

}

}

    umbilical.done(getTaskID()); 獲取更多大數(shù)據(jù)視頻資料請加QQ群：947967114

//實(shí)際上這里向MRAppMaster上的TaskAttemptImpl發(fā)送TA_DONE事件，在TA_DONE事件的驅(qū)動下，相應(yīng)的TaskAttemptImpl對象的狀態(tài)機(jī)執(zhí)行CleanupContainerTransition.transition，然后轉(zhuǎn)入SUCCESS_CONTAINER_CLEANUP狀態(tài)。注意這里有一個TaskAttemptEventType.TA_DONE事件是由具體的MapTask所在節(jié)點(diǎn)上發(fā)出的，但不是引起的狀態(tài)機(jī)的跳變是在MRAppMaster節(jié)點(diǎn)上。對于Maptask，會有一個umbilical，就代表著MRAppMaster。

MPAppmaster接到CONTAINER_REMOTE_CLEANUP事件，ContainerLauncher通過RPC機(jī)制調(diào)用Maptask所在節(jié)點(diǎn)的ContainerManagerImpl.stopContainers.使這個MapTask的容器進(jìn)入KILLED_BY_APPMASTER狀態(tài)從而不在活躍。操作成功后向相應(yīng)的TaskAttemptImpl發(fā)送TO_CONTAINER_CLEANED事件。如果一次TaskAttempt成功了，就意味著嘗試的任務(wù)也成功了，所以TaskAttempt的狀態(tài)關(guān)系到TaskImpl對象，taskImpl的掃描和善后，包括向上層的JobImpl對象發(fā)送TaskState.SUCCESSED事件。向自身TaskImpl發(fā)送的SUCCESSED事件會導(dǎo)致TaskImpl.handleTaskAttemptCompletion操作。

Mapper節(jié)點(diǎn)上產(chǎn)生一個過程setMapOutputServerAdress函數(shù)，把本節(jié)點(diǎn)的MapOutputServer地址設(shè)置成一個Web地址，意味著MapTask留下的數(shù)據(jù)輸出（合并后的spill文件）可以通過HTTP連接獲取。至此Mapper的所有過程完成。 獲取更多大數(shù)據(jù)視頻資料請加QQ群：947967114