Hadoop中的map，shuffle，merge和减少时间的精确定义

Question

在Hadoop中，作业执行后会提供以下指标：

• 地图时间
• 减少时间
• 洗牌时间
• 合并时间

我找不到这些时间的确切定义，因为所有来源仍不清楚如何精确计算这些时间。这是我的看法：

• 地图时间是读取输入，应用地图功能和对数据进行排序的时间
• 减少时间是应用减少功能并写入输出的时间
• 随机播放时间是指将地图分类的数据合并到转移到还原器的时间
• 合并时间是仅在缩减端合并映射输出的时间

我不确定黑体字的内容。我的分析正确吗？

Answer 1

我决定研究Hadoop代码以获得更多见解。下图说明了我的发现。

我发现：

• 映射时间是映射任务花费的时间。地图任务负责读取输入，应用地图功能，对数据进行排序和合并数据。
• 随机播放时间是将地图输出数据复制到reduce任务的时间，这是reduce任务的一部分。
• 合并时间是在归约端合并映射输出的时间，这是归约任务的一部分。
• 减少时间是应用减少功能并写入输出的时间。

以下代码支持这些发现：

在ReduceTask使用的Shuffle类中，我们看到“复制”阶段之后是“排序”阶段。

copyPhase.complete(); // copy is already complete
taskStatus.setPhase(TaskStatus.Phase.SORT);
reduceTask.statusUpdate(umbilical);

// Finish the on-going merges...
RawKeyValueIterator kvIter = null;
try {
  kvIter = merger.close();
} catch (Throwable e) {
  throw new ShuffleError("Error while doing final merge " , e);
}

在TaskStatus类中，我们看到shuffletime是排序阶段之前的时间，sort time是shuffle和reduce阶段之间的时间。

public void setPhase(Phase phase){
  TaskStatus.Phase oldPhase = getPhase();
  if (oldPhase != phase){
    // sort phase started
    if (phase == TaskStatus.Phase.SORT){
      if (oldPhase == TaskStatus.Phase.MAP) {
        setMapFinishTime(System.currentTimeMillis());
      }
      else {
        setShuffleFinishTime(System.currentTimeMillis());
      }
    }else if (phase == TaskStatus.Phase.REDUCE){
      setSortFinishTime(System.currentTimeMillis());
    }
    this.phase = phase;
  }
  ...

在JobInfo类中，我们看到混洗时间与复制相对应，并且合并时间是我们上面提到的“排序”时间。

switch (task.getType()) {
    case MAP:
      successfulMapAttempts += successful;
      failedMapAttempts += failed;
      killedMapAttempts += killed;
      if (attempt.getState() == TaskAttemptState.SUCCEEDED) {
        numMaps++;
        avgMapTime += (attempt.getFinishTime() - attempt.getLaunchTime());
      }
      break;
    case REDUCE:
      successfulReduceAttempts += successful;
      failedReduceAttempts += failed;
      killedReduceAttempts += killed;
      if (attempt.getState() == TaskAttemptState.SUCCEEDED) {
        numReduces++;
        avgShuffleTime += (attempt.getShuffleFinishTime() - attempt
            .getLaunchTime());
        avgMergeTime += attempt.getSortFinishTime()
            - attempt.getShuffleFinishTime();
        avgReduceTime += (attempt.getFinishTime() - attempt
            .getSortFinishTime());
      }
}

有关reduce和map任务如何工作的更多信息可以分别从MapTask和ReduceTask类派生。

最后，我想指出的是，我在链接中引用的源代码大部分对应于Hadoop 2.7.1代码。