hadoop - Hadoop 中 map、shuffle、merge 和 reduce 时间的精确定义

Question

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

• map 时间
• 减少时间
• 洗牌时间
• 合并时间

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

• map 时间是读取输入和应用 map 功能以及排序数据
的时间
• reduce 时间是应用reduce 函数并编写输出
的时间
• Shuffle time 是merge map 排序数据传输到reducer 的时间
• 合并时间是 仅在 reduce 端合并 map 输出的时间

我不确定粗体部分。我的分析正确吗？

Answer 1

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

我发现:

• map 时间是 map task 所花费的时间。 Map任务负责读取输入、应用map函数、排序数据和合并数据。
• Shuffle time 是将 map 输出数据复制到 reduce 任务的时间，这是 reduce 任务的一部分。
• 合并时间是在 reduce 端合并 map 输出的时间，这是 reduce 任务的一部分。
• 减少时间是应用减少功能并写入输出的时间。

这些发现得到以下代码的支持:

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

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 是排序阶段之前的时间，排序时间是 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 class我们看到shuffle时间对应的是copy，merge时间就是我们上面说的“sort”时间。

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 代码。