[记录点滴] 小心 Hadoop Speculative 调度策略

int main(int argc, char** argv) {
    string key;
    while(cin >> key) {
        key是文件名字，
        解析这个大文件为若干小文件，然后存储
        cout << key << "/t" << "1" << endl;
    }
    return 0;
}

int main(int argc, char** argv) {
    string key, num;
    map<string, int> count; 
    map<string, int>::iterator it;
  
    while(cin >> key >> num) {
        it = count.find(key);
        if(it != count.end()) {
            it->second++;
        }
        else {
            count.insert(make_pair(key, 1));
        }
    }

    for(it = count.begin(); it != count.end(); it++) {
        cout << it->first << "/t" << it->second << endl;
    }
    return 0;
}

<property>
  <name>mapreduce.map.speculative</name>
  <value>false</value>
</property>

<property>
  <name>mapreduce.reduce.speculative</name>
  <value>false</value>
</property>

private int maybeScheduleASpeculation(TaskType type) {
    int successes = 0;
    long now = clock.getTime();
    //根据当前Task的类型（map或reduce）获取相应类型任务的需要分配Container数量的缓存containerNeeds
    ConcurrentMap<JobId, AtomicInteger> containerNeeds
        = type == TaskType.MAP ? mapContainerNeeds : reduceContainerNeeds;

    //遍历containerNeeds
    for (ConcurrentMap.Entry<JobId, AtomicInteger> jobEntry : containerNeeds.entrySet()) {
      // This race conditon is okay.  If we skip a speculation attempt we
      //  should have tried because the event that lowers the number of
      //  containers needed to zero hasn't come through, it will next time.
      // Also, if we miss the fact that the number of containers needed was
      //  zero but increased due to a failure it's not too bad to launch one
      //  container prematurely.
      //如果当前Job依然有未分配Container的Task，那么跳过当前循环，继续下一次循环。这说明如果当前Job的某一类型的Task依然存在未分配Container的，则不会进行任务推断；
      if (jobEntry.getValue().get() > 0) {
        continue;
      }

      int numberSpeculationsAlready = 0;
      int numberRunningTasks = 0;

      // loop through the tasks of the kind
      //从当前应用的上下文AppContext中获取Job，并获取此Job的所有的Task（map或者reduce）
      Job job = context.getJob(jobEntry.getKey());
      Map<TaskId, Task> tasks = job.getTasks(type);

      //计算允许执行推断的Task数量numberAllowedSpeculativeTasks（map或者reduce）
      int numberAllowedSpeculativeTasks
          = (int) Math.max(minimumAllowedSpeculativeTasks,
              proportionTotalTasksSpeculatable * tasks.size());

      TaskId bestTaskID = null;
      long bestSpeculationValue = -1L;

      // this loop is potentially pricey.
      // TODO track the tasks that are potentially worth looking at
      //遍历Job对应的map任务或者reduce任务集合，在迭代完所有的map任务或者reduce任务后，获取这一任务集合中的推断值bestSpeculationValue最大的任务ID。
      for (Map.Entry<TaskId, Task> taskEntry : tasks.entrySet()) {
        //调用speculationValue方法获取每一个Task的推断值。
        long mySpeculationValue = speculationValue(taskEntry.getKey(), now);

        if (mySpeculationValue == ALREADY_SPECULATING) {
          ++numberSpeculationsAlready;
        }

        if (mySpeculationValue != NOT_RUNNING) {
          ++numberRunningTasks;
        }

        //获取这一任务集合中的推断值bestSpeculationValue最大的任务ID
        if (mySpeculationValue > bestSpeculationValue) {
          bestTaskID = taskEntry.getKey();
          bestSpeculationValue = mySpeculationValue;
        }
      }
      //再次计算numberAllowedSpeculativeTasks
      numberAllowedSpeculativeTasks
          = (int) Math.max(numberAllowedSpeculativeTasks,
              proportionRunningTasksSpeculatable * numberRunningTasks);

      // If we found a speculation target, fire it off
      //如果numberAllowedSpeculativeTasks大于numberSpeculationsAlready（已经推断执行过的Task数量），则调用addSpeculativeAttempt方法将第4步中选出的任务的任务ID添加到推断尝试中。
      if (bestTaskID != null
          && numberAllowedSpeculativeTasks > numberSpeculationsAlready) {
        addSpeculativeAttempt(bestTaskID);
        ++successes;
      }
    }

    return successes;
}

private long speculationValue(TaskId taskID, long now) {
    Job job = context.getJob(taskID.getJobId());
    Task task = job.getTask(taskID);
    Map<TaskAttemptId, TaskAttempt> attempts = task.getAttempts();
    long acceptableRuntime = Long.MIN_VALUE;
    long result = Long.MIN_VALUE;

    //如果任务还没有被推断执行，那么调用estimator的thresholdRuntime方法获取任务可以接受的运行时长acceptableRuntime
    if (!mayHaveSpeculated.contains(taskID)) {
      acceptableRuntime = estimator.thresholdRuntime(taskID);
      if (acceptableRuntime == Long.MAX_VALUE) {
        return ON_SCHEDULE;
      }
    }

    TaskAttemptId runningTaskAttemptID = null;

    int numberRunningAttempts = 0;

    for (TaskAttempt taskAttempt : attempts.values()) {
      if (taskAttempt.getState() == TaskAttemptState.RUNNING
          || taskAttempt.getState() == TaskAttemptState.STARTING) {
        //如果任务的运行实例数大于1，则说明此任务已经发生了推断执行，因此返回ALREADY_SPECULATING。
        if (++numberRunningAttempts > 1) {
          return ALREADY_SPECULATING;
        }
        runningTaskAttemptID = taskAttempt.getID();

        //调用estimator的estimatedRuntime方法获取任务运行实例的估算运行时长estimatedRunTime。
        long estimatedRunTime = estimator.estimatedRuntime(runningTaskAttemptID);

        //调用estimator的attemptEnrolledTime方法获取任务实例开始运行的时间，此时间即为startTimes中缓存的start。
        long taskAttemptStartTime
            = estimator.attemptEnrolledTime(runningTaskAttemptID);
        if (taskAttemptStartTime > now) {
          // This background process ran before we could process the task
          //  attempt status change that chronicles the attempt start
          return TOO_NEW;
        }

        //estimatedEndTime表示估算任务实例的运行结束时间
        long estimatedEndTime = estimatedRunTime + taskAttemptStartTime;

        //调用estimator的estimatedNewAttemptRuntime方法估算如果此时重新为任务启动一个实例，此实例运行结束的时间estimatedReplacementEndTime。
        long estimatedReplacementEndTime
            = now + estimator.estimatedNewAttemptRuntime(taskID);

        float progress = taskAttempt.getProgress();
        TaskAttemptHistoryStatistics data =
            runningTaskAttemptStatistics.get(runningTaskAttemptID);
        if (data == null) {
          //如果缓存中没有任务实例的历史统计信息，那么将estimatedRunTime、任务实例进度progress，当前时间封装为历史统计信息缓存起来。
          runningTaskAttemptStatistics.put(runningTaskAttemptID,
            new TaskAttemptHistoryStatistics(estimatedRunTime, progress, now));
        } else {
          //如果缓存中存在任务实例的历史统计信息
          if (estimatedRunTime == data.getEstimatedRunTime()
              && progress == data.getProgress()) {
            //如果缓存的estimatedRunTime和本次估算的estimatedRunTime一样并且缓存的实例进度progress和本次获取的任务实例进度progress一样
            // Previous stats are same as same stats
            if (data.notHeartbeatedInAWhile(now)
                || estimator.hasStagnatedProgress(runningTaskAttemptID, now)) {
              // Stats have stagnated for a while, simulate heart-beat.
              TaskAttemptStatus taskAttemptStatus = new TaskAttemptStatus();
              taskAttemptStatus.id = runningTaskAttemptID;
              taskAttemptStatus.progress = progress;
              taskAttemptStatus.taskState = taskAttempt.getState();
              // Now simulate the heart-beat
              //说明有一段时间没有收到心跳了，则模拟一次心跳。
              handleAttempt(taskAttemptStatus);
            }
          } else {
            //如果缓存的estimatedRunTime和本次估算的estimatedRunTime不一样或者缓存的实例进度progress和本次获取的任务实例进度progress不一样
            // Stats have changed - update our data structure
            //将estimatedRunTime、任务实例进度progress，当前时间更新到任务实例的历史统计信息中。
            data.setEstimatedRunTime(estimatedRunTime);
            data.setProgress(progress);
            data.resetHeartBeatTime(now);
          }
        }

        //如果estimatedEndTime小于当前时间，则说明任务实例的进度良好，返回PROGRESS_IS_GOOD，PROGRESS_IS_GOOD等于Long.MIN_VALUE + 3。
        if (estimatedEndTime < now) {
          return PROGRESS_IS_GOOD;
        }

        //如果estimatedReplacementEndTime大于等于estimatedEndTime，则说明即便启动备份任务实例也无济于事，因为它的结束时间达不到节省作业总运行时长的作用。
        if (estimatedReplacementEndTime >= estimatedEndTime) {
          return TOO_LATE_TO_SPECULATE;
        }

        //计算本次估算的结果值result，它等于estimatedEndTime - estimatedReplacementEndTime，当这个差值越大表示备份任务实例运行后比原任务实例的结束时间就越早，因此调度执行的价值越大。
        result = estimatedEndTime - estimatedReplacementEndTime;
      }
    }

    // If we are here, there's at most one task attempt.
    if (numberRunningAttempts == 0) {
      //如果numberRunningAttempts等于0，则表示当前任务还没有启动任务实例，返回NOT_RUNNING，NOT_RUNNING等于Long.MIN_VALUE + 4。
      return NOT_RUNNING;
    }

    //重新计算acceptableRuntime，处理方式与第1步相同。
    if (acceptableRuntime == Long.MIN_VALUE) {
      acceptableRuntime = estimator.thresholdRuntime(taskID);
      if (acceptableRuntime == Long.MAX_VALUE) {
        return ON_SCHEDULE;
      }
    }

    return result;
}