org.apache.hadoop.mapreduce.JobCounter源码实例Demo-ICode联盟

类org.apache.hadoop.mapreduce.JobCounter源码实例Demo

下面列出了怎么用org.apache.hadoop.mapreduce.JobCounter的API类实例代码及写法，或者点击链接到github查看源代码。

源代码1 项目： hadoop 文件： TaskAttemptImpl.java

@SuppressWarnings("unchecked")
private void sendLaunchedEvents() {
  JobCounterUpdateEvent jce = new JobCounterUpdateEvent(attemptId.getTaskId()
      .getJobId());
  jce.addCounterUpdate(attemptId.getTaskId().getTaskType() == TaskType.MAP ?
      JobCounter.TOTAL_LAUNCHED_MAPS : JobCounter.TOTAL_LAUNCHED_REDUCES, 1);
  eventHandler.handle(jce);

  LOG.info("TaskAttempt: [" + attemptId
      + "] using containerId: [" + container.getId() + " on NM: ["
      + StringInterner.weakIntern(container.getNodeId().toString()) + "]");
  TaskAttemptStartedEvent tase =
    new TaskAttemptStartedEvent(TypeConverter.fromYarn(attemptId),
        TypeConverter.fromYarn(attemptId.getTaskId().getTaskType()),
        launchTime, trackerName, httpPort, shufflePort, container.getId(),
        locality.toString(), avataar.toString());
  eventHandler.handle(
      new JobHistoryEvent(attemptId.getTaskId().getJobId(), tase));
}

源代码2 项目： hadoop 文件： RMContainerAllocator.java

@SuppressWarnings("unchecked")
private ContainerRequest assignToFailedMap(Container allocated) {
  //try to assign to earlierFailedMaps if present
  ContainerRequest assigned = null;
  while (assigned == null && earlierFailedMaps.size() > 0
      && canAssignMaps()) {
    TaskAttemptId tId = earlierFailedMaps.removeFirst();      
    if (maps.containsKey(tId)) {
      assigned = maps.remove(tId);
      JobCounterUpdateEvent jce =
        new JobCounterUpdateEvent(assigned.attemptID.getTaskId().getJobId());
      jce.addCounterUpdate(JobCounter.OTHER_LOCAL_MAPS, 1);
      eventHandler.handle(jce);
      LOG.info("Assigned from earlierFailedMaps");
      break;
    }
  }
  return assigned;
}

源代码3 项目： hadoop 文件： TestMultipleLevelCaching.java

/**
 * Launches a MR job and tests the job counters against the expected values.
 * @param testName The name for the job
 * @param mr The MR cluster
 * @param fileSys The FileSystem
 * @param in Input path
 * @param out Output path
 * @param numMaps Number of maps
 * @param otherLocalMaps Expected value of other local maps
 * @param datalocalMaps Expected value of data(node) local maps
 * @param racklocalMaps Expected value of rack local maps
 */
static void launchJobAndTestCounters(String jobName, MiniMRCluster mr,
                                     FileSystem fileSys, Path in, Path out,
                                     int numMaps, int otherLocalMaps,
                                     int dataLocalMaps, int rackLocalMaps)
throws IOException {
  JobConf jobConf = mr.createJobConf();
  if (fileSys.exists(out)) {
      fileSys.delete(out, true);
  }
  RunningJob job = launchJob(jobConf, in, out, numMaps, jobName);
  Counters counters = job.getCounters();
  assertEquals("Number of local maps",
          counters.getCounter(JobCounter.OTHER_LOCAL_MAPS), otherLocalMaps);
  assertEquals("Number of Data-local maps",
          counters.getCounter(JobCounter.DATA_LOCAL_MAPS),
                              dataLocalMaps);
  assertEquals("Number of Rack-local maps",
          counters.getCounter(JobCounter.RACK_LOCAL_MAPS),
                              rackLocalMaps);
  mr.waitUntilIdle();
  mr.shutdown();
}

源代码4 项目： hadoop 文件： TestMRJobs.java

protected void verifySleepJobCounters(Job job) throws InterruptedException,
    IOException {
  Counters counters = job.getCounters();
  Assert.assertEquals(3, counters.findCounter(JobCounter.OTHER_LOCAL_MAPS)
      .getValue());
  Assert.assertEquals(3, counters.findCounter(JobCounter.TOTAL_LAUNCHED_MAPS)
      .getValue());
  Assert.assertEquals(numSleepReducers,
      counters.findCounter(JobCounter.TOTAL_LAUNCHED_REDUCES).getValue());
  Assert
      .assertTrue(counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS) != null
          && counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS).getValue() != 0);
  Assert
      .assertTrue(counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS) != null
          && counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS).getValue() != 0);
}

源代码5 项目： hadoop 文件： TestCounters.java

@SuppressWarnings("deprecation")
private void checkLegacyNames(Counters counters) {
  assertEquals("New name", 1, counters.findCounter(
      TaskCounter.class.getName(), "MAP_INPUT_RECORDS").getValue());
  assertEquals("Legacy name", 1, counters.findCounter(
      "org.apache.hadoop.mapred.Task$Counter",
      "MAP_INPUT_RECORDS").getValue());
  assertEquals("Legacy enum", 1,
      counters.findCounter(Task.Counter.MAP_INPUT_RECORDS).getValue());

  assertEquals("New name", 1, counters.findCounter(
      JobCounter.class.getName(), "DATA_LOCAL_MAPS").getValue());
  assertEquals("Legacy name", 1, counters.findCounter(
      "org.apache.hadoop.mapred.JobInProgress$Counter",
      "DATA_LOCAL_MAPS").getValue());
  assertEquals("Legacy enum", 1,
      counters.findCounter(JobInProgress.Counter.DATA_LOCAL_MAPS).getValue());

  assertEquals("New name", 1, counters.findCounter(
      FileSystemCounter.class.getName(), "FILE_BYTES_READ").getValue());
  assertEquals("New name and method", 1, counters.findCounter("file",
      FileSystemCounter.BYTES_READ).getValue());
  assertEquals("Legacy name", 1, counters.findCounter(
      "FileSystemCounters",
      "FILE_BYTES_READ").getValue());
}

源代码6 项目： hadoop 文件： TestCounters.java

@SuppressWarnings("rawtypes")
@Test
public void testFrameworkCounter() {
  GroupFactory groupFactory = new GroupFactoryForTest();
  FrameworkGroupFactory frameworkGroupFactory = 
      groupFactory.newFrameworkGroupFactory(JobCounter.class);
  Group group = (Group) frameworkGroupFactory.newGroup("JobCounter");
  
  FrameworkCounterGroup counterGroup = 
      (FrameworkCounterGroup) group.getUnderlyingGroup();

  org.apache.hadoop.mapreduce.Counter count1 = 
      counterGroup.findCounter(JobCounter.NUM_FAILED_MAPS.toString());
  Assert.assertNotNull(count1);
  
  // Verify no exception get thrown when finding an unknown counter
  org.apache.hadoop.mapreduce.Counter count2 = 
      counterGroup.findCounter("Unknown");
  Assert.assertNull(count2);
}

源代码7 项目： big-c 文件： TaskAttemptImpl.java

@SuppressWarnings("unchecked")
private void sendLaunchedEvents() {
  JobCounterUpdateEvent jce = new JobCounterUpdateEvent(attemptId.getTaskId()
      .getJobId());
  jce.addCounterUpdate(attemptId.getTaskId().getTaskType() == TaskType.MAP ?
      JobCounter.TOTAL_LAUNCHED_MAPS : JobCounter.TOTAL_LAUNCHED_REDUCES, 1);
  eventHandler.handle(jce);

  LOG.info("TaskAttempt: [" + attemptId
      + "] using containerId: [" + container.getId() + " on NM: ["
      + StringInterner.weakIntern(container.getNodeId().toString()) + "]");
  TaskAttemptStartedEvent tase =
    new TaskAttemptStartedEvent(TypeConverter.fromYarn(attemptId),
        TypeConverter.fromYarn(attemptId.getTaskId().getTaskType()),
        launchTime, trackerName, httpPort, shufflePort, container.getId(),
        locality.toString(), avataar.toString());
  eventHandler.handle(
      new JobHistoryEvent(attemptId.getTaskId().getJobId(), tase));
}

源代码8 项目： big-c 文件： RMContainerAllocator.java

@SuppressWarnings("unchecked")
private ContainerRequest assignToFailedMap(Container allocated) {
  //try to assign to earlierFailedMaps if present
  ContainerRequest assigned = null;
  while (assigned == null && earlierFailedMaps.size() > 0
      && canAssignMaps()) {
    TaskAttemptId tId = earlierFailedMaps.removeFirst();      
    if (maps.containsKey(tId)) {
      assigned = maps.remove(tId);
      JobCounterUpdateEvent jce =
        new JobCounterUpdateEvent(assigned.attemptID.getTaskId().getJobId());
      jce.addCounterUpdate(JobCounter.OTHER_LOCAL_MAPS, 1);
      eventHandler.handle(jce);
      LOG.info("Assigned from earlierFailedMaps");
      break;
    }
  }
  return assigned;
}

源代码9 项目： big-c 文件： TestMultipleLevelCaching.java

/**
 * Launches a MR job and tests the job counters against the expected values.
 * @param testName The name for the job
 * @param mr The MR cluster
 * @param fileSys The FileSystem
 * @param in Input path
 * @param out Output path
 * @param numMaps Number of maps
 * @param otherLocalMaps Expected value of other local maps
 * @param datalocalMaps Expected value of data(node) local maps
 * @param racklocalMaps Expected value of rack local maps
 */
static void launchJobAndTestCounters(String jobName, MiniMRCluster mr,
                                     FileSystem fileSys, Path in, Path out,
                                     int numMaps, int otherLocalMaps,
                                     int dataLocalMaps, int rackLocalMaps)
throws IOException {
  JobConf jobConf = mr.createJobConf();
  if (fileSys.exists(out)) {
      fileSys.delete(out, true);
  }
  RunningJob job = launchJob(jobConf, in, out, numMaps, jobName);
  Counters counters = job.getCounters();
  assertEquals("Number of local maps",
          counters.getCounter(JobCounter.OTHER_LOCAL_MAPS), otherLocalMaps);
  assertEquals("Number of Data-local maps",
          counters.getCounter(JobCounter.DATA_LOCAL_MAPS),
                              dataLocalMaps);
  assertEquals("Number of Rack-local maps",
          counters.getCounter(JobCounter.RACK_LOCAL_MAPS),
                              rackLocalMaps);
  mr.waitUntilIdle();
  mr.shutdown();
}

源代码10 项目： big-c 文件： TestMRJobs.java

protected void verifySleepJobCounters(Job job) throws InterruptedException,
    IOException {
  Counters counters = job.getCounters();
  Assert.assertEquals(3, counters.findCounter(JobCounter.OTHER_LOCAL_MAPS)
      .getValue());
  Assert.assertEquals(3, counters.findCounter(JobCounter.TOTAL_LAUNCHED_MAPS)
      .getValue());
  Assert.assertEquals(numSleepReducers,
      counters.findCounter(JobCounter.TOTAL_LAUNCHED_REDUCES).getValue());
  Assert
      .assertTrue(counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS) != null
          && counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS).getValue() != 0);
  Assert
      .assertTrue(counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS) != null
          && counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS).getValue() != 0);
}

源代码11 项目： big-c 文件： TestCounters.java

@SuppressWarnings("deprecation")
private void checkLegacyNames(Counters counters) {
  assertEquals("New name", 1, counters.findCounter(
      TaskCounter.class.getName(), "MAP_INPUT_RECORDS").getValue());
  assertEquals("Legacy name", 1, counters.findCounter(
      "org.apache.hadoop.mapred.Task$Counter",
      "MAP_INPUT_RECORDS").getValue());
  assertEquals("Legacy enum", 1,
      counters.findCounter(Task.Counter.MAP_INPUT_RECORDS).getValue());

  assertEquals("New name", 1, counters.findCounter(
      JobCounter.class.getName(), "DATA_LOCAL_MAPS").getValue());
  assertEquals("Legacy name", 1, counters.findCounter(
      "org.apache.hadoop.mapred.JobInProgress$Counter",
      "DATA_LOCAL_MAPS").getValue());
  assertEquals("Legacy enum", 1,
      counters.findCounter(JobInProgress.Counter.DATA_LOCAL_MAPS).getValue());

  assertEquals("New name", 1, counters.findCounter(
      FileSystemCounter.class.getName(), "FILE_BYTES_READ").getValue());
  assertEquals("New name and method", 1, counters.findCounter("file",
      FileSystemCounter.BYTES_READ).getValue());
  assertEquals("Legacy name", 1, counters.findCounter(
      "FileSystemCounters",
      "FILE_BYTES_READ").getValue());
}

源代码12 项目： big-c 文件： TestCounters.java

@SuppressWarnings("rawtypes")
@Test
public void testFrameworkCounter() {
  GroupFactory groupFactory = new GroupFactoryForTest();
  FrameworkGroupFactory frameworkGroupFactory = 
      groupFactory.newFrameworkGroupFactory(JobCounter.class);
  Group group = (Group) frameworkGroupFactory.newGroup("JobCounter");
  
  FrameworkCounterGroup counterGroup = 
      (FrameworkCounterGroup) group.getUnderlyingGroup();

  org.apache.hadoop.mapreduce.Counter count1 = 
      counterGroup.findCounter(JobCounter.NUM_FAILED_MAPS.toString());
  Assert.assertNotNull(count1);
  
  // Verify no exception get thrown when finding an unknown counter
  org.apache.hadoop.mapreduce.Counter count2 = 
      counterGroup.findCounter("Unknown");
  Assert.assertNull(count2);
}

源代码13 项目： hadoop 文件： LocalContainerAllocator.java

@SuppressWarnings("unchecked")
@Override
public void handle(ContainerAllocatorEvent event) {
  if (event.getType() == ContainerAllocator.EventType.CONTAINER_REQ) {
    LOG.info("Processing the event " + event.toString());
    // Assign the same container ID as the AM
    ContainerId cID =
        ContainerId.newContainerId(getContext().getApplicationAttemptId(),
          this.containerId.getContainerId());
    Container container = recordFactory.newRecordInstance(Container.class);
    container.setId(cID);
    NodeId nodeId = NodeId.newInstance(this.nmHost, this.nmPort);
    container.setNodeId(nodeId);
    container.setContainerToken(null);
    container.setNodeHttpAddress(this.nmHost + ":" + this.nmHttpPort);
    // send the container-assigned event to task attempt

    if (event.getAttemptID().getTaskId().getTaskType() == TaskType.MAP) {
      JobCounterUpdateEvent jce =
          new JobCounterUpdateEvent(event.getAttemptID().getTaskId()
              .getJobId());
      // TODO Setting OTHER_LOCAL_MAP for now.
      jce.addCounterUpdate(JobCounter.OTHER_LOCAL_MAPS, 1);
      eventHandler.handle(jce);
    }
    eventHandler.handle(new TaskAttemptContainerAssignedEvent(
        event.getAttemptID(), container, applicationACLs));
  }
}

源代码14 项目： hadoop 文件： TaskAttemptImpl.java

private static void updateMillisCounters(JobCounterUpdateEvent jce,
    TaskAttemptImpl taskAttempt) {
  TaskType taskType = taskAttempt.getID().getTaskId().getTaskType();
  long duration = (taskAttempt.getFinishTime() - taskAttempt.getLaunchTime());
  int mbRequired =
      taskAttempt.getMemoryRequired(taskAttempt.conf, taskType);
  int vcoresRequired = taskAttempt.getCpuRequired(taskAttempt.conf, taskType);
  int gcoresRequired = taskAttempt.getGpuRequired(taskAttempt.conf, taskType);

  int minSlotMemSize = taskAttempt.conf.getInt(
    YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB,
    YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB);

  int simSlotsRequired =
      minSlotMemSize == 0 ? 0 : (int) Math.ceil((float) mbRequired
          / minSlotMemSize);

  if (taskType == TaskType.MAP) {
    jce.addCounterUpdate(JobCounter.SLOTS_MILLIS_MAPS, simSlotsRequired * duration);
    jce.addCounterUpdate(JobCounter.MB_MILLIS_MAPS, duration * mbRequired);
    jce.addCounterUpdate(JobCounter.VCORES_MILLIS_MAPS, duration * vcoresRequired);
    jce.addCounterUpdate(JobCounter.GCORES_MILLIS_MAPS, duration * gcoresRequired);
    jce.addCounterUpdate(JobCounter.MILLIS_MAPS, duration);
  } else {
    jce.addCounterUpdate(JobCounter.SLOTS_MILLIS_REDUCES, simSlotsRequired * duration);
    jce.addCounterUpdate(JobCounter.MB_MILLIS_REDUCES, duration * mbRequired);
    jce.addCounterUpdate(JobCounter.VCORES_MILLIS_REDUCES, duration * vcoresRequired);
    jce.addCounterUpdate(JobCounter.GCORES_MILLIS_REDUCES, duration * gcoresRequired);
    jce.addCounterUpdate(JobCounter.MILLIS_REDUCES, duration);
  }
}

源代码15 项目： hadoop 文件： TaskAttemptImpl.java

private static JobCounterUpdateEvent createJobCounterUpdateEventTAFailed(
    TaskAttemptImpl taskAttempt, boolean taskAlreadyCompleted) {
  TaskType taskType = taskAttempt.getID().getTaskId().getTaskType();
  JobCounterUpdateEvent jce = new JobCounterUpdateEvent(taskAttempt.getID().getTaskId().getJobId());
  
  if (taskType == TaskType.MAP) {
    jce.addCounterUpdate(JobCounter.NUM_FAILED_MAPS, 1);
  } else {
    jce.addCounterUpdate(JobCounter.NUM_FAILED_REDUCES, 1);
  }
  if (!taskAlreadyCompleted) {
    updateMillisCounters(jce, taskAttempt);
  }
  return jce;
}

源代码16 项目： hadoop 文件： TaskAttemptImpl.java

private static JobCounterUpdateEvent createJobCounterUpdateEventTAKilled(
    TaskAttemptImpl taskAttempt, boolean taskAlreadyCompleted) {
  TaskType taskType = taskAttempt.getID().getTaskId().getTaskType();
  JobCounterUpdateEvent jce = new JobCounterUpdateEvent(taskAttempt.getID().getTaskId().getJobId());
  
  if (taskType == TaskType.MAP) {
    jce.addCounterUpdate(JobCounter.NUM_KILLED_MAPS, 1);
  } else {
    jce.addCounterUpdate(JobCounter.NUM_KILLED_REDUCES, 1);
  }
  if (!taskAlreadyCompleted) {
    updateMillisCounters(jce, taskAttempt);
  }
  return jce;
}

源代码17 项目： hadoop 文件： JobHistoryEventHandler.java

private void setSummarySlotSeconds(JobSummary summary, Counters allCounters) {

    Counter slotMillisMapCounter = allCounters
      .findCounter(JobCounter.SLOTS_MILLIS_MAPS);
    if (slotMillisMapCounter != null) {
      summary.setMapSlotSeconds(slotMillisMapCounter.getValue() / 1000);
    }

    Counter slotMillisReduceCounter = allCounters
      .findCounter(JobCounter.SLOTS_MILLIS_REDUCES);
    if (slotMillisReduceCounter != null) {
      summary.setReduceSlotSeconds(slotMillisReduceCounter.getValue() / 1000);
    }
  }

源代码18 项目： hadoop 文件： TestMRJobs.java

protected void verifyRandomWriterCounters(Job job)
    throws InterruptedException, IOException {
  Counters counters = job.getCounters();
  Assert.assertEquals(3, counters.findCounter(JobCounter.OTHER_LOCAL_MAPS)
      .getValue());
  Assert.assertEquals(3, counters.findCounter(JobCounter.TOTAL_LAUNCHED_MAPS)
      .getValue());
  Assert
      .assertTrue(counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS) != null
          && counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS).getValue() != 0);
}

源代码19 项目： hadoop 文件： TestMRJobs.java

protected void verifyFailingMapperCounters(Job job)
    throws InterruptedException, IOException {
  Counters counters = job.getCounters();
  Assert.assertEquals(2, counters.findCounter(JobCounter.OTHER_LOCAL_MAPS)
      .getValue());
  Assert.assertEquals(2, counters.findCounter(JobCounter.TOTAL_LAUNCHED_MAPS)
      .getValue());
  Assert.assertEquals(2, counters.findCounter(JobCounter.NUM_FAILED_MAPS)
      .getValue());
  Assert
      .assertTrue(counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS) != null
          && counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS).getValue() != 0);
}

源代码20 项目： hadoop 文件： TestUberAM.java

@Override
protected void verifySleepJobCounters(Job job) throws InterruptedException,
    IOException {
  Counters counters = job.getCounters();
  super.verifySleepJobCounters(job);
  Assert.assertEquals(3,
      counters.findCounter(JobCounter.NUM_UBER_SUBMAPS).getValue());
  Assert.assertEquals(numSleepReducers,
      counters.findCounter(JobCounter.NUM_UBER_SUBREDUCES).getValue());
  Assert.assertEquals(3 + numSleepReducers,
      counters.findCounter(JobCounter.TOTAL_LAUNCHED_UBERTASKS).getValue());
}

源代码21 项目： hadoop 文件： TestUberAM.java

@Override
protected void verifyRandomWriterCounters(Job job)
    throws InterruptedException, IOException {
  super.verifyRandomWriterCounters(job);
  Counters counters = job.getCounters();
  Assert.assertEquals(3, counters.findCounter(JobCounter.NUM_UBER_SUBMAPS)
      .getValue());
  Assert.assertEquals(3,
      counters.findCounter(JobCounter.TOTAL_LAUNCHED_UBERTASKS).getValue());
}

源代码22 项目： hadoop 文件： TestUberAM.java

@Override
protected void verifyFailingMapperCounters(Job job)
    throws InterruptedException, IOException {
  Counters counters = job.getCounters();
  super.verifyFailingMapperCounters(job);
  Assert.assertEquals(2,
      counters.findCounter(JobCounter.TOTAL_LAUNCHED_UBERTASKS).getValue());
  Assert.assertEquals(2, counters.findCounter(JobCounter.NUM_UBER_SUBMAPS)
      .getValue());
  Assert.assertEquals(2, counters
      .findCounter(JobCounter.NUM_FAILED_UBERTASKS).getValue());
}

源代码23 项目： hadoop 文件： TestCounters.java

@SuppressWarnings("deprecation")
@Test
public void testReadWithLegacyNames() {
  Counters counters = new Counters();
  counters.incrCounter(TaskCounter.MAP_INPUT_RECORDS, 1);
  counters.incrCounter(JobCounter.DATA_LOCAL_MAPS, 1);
  counters.findCounter("file", FileSystemCounter.BYTES_READ).increment(1);
  
  checkLegacyNames(counters);
}

源代码24 项目： big-c 文件： LocalContainerAllocator.java

@SuppressWarnings("unchecked")
@Override
public void handle(ContainerAllocatorEvent event) {
  if (event.getType() == ContainerAllocator.EventType.CONTAINER_REQ) {
    LOG.info("Processing the event " + event.toString());
    // Assign the same container ID as the AM
    ContainerId cID =
        ContainerId.newContainerId(getContext().getApplicationAttemptId(),
          this.containerId.getContainerId());
    Container container = recordFactory.newRecordInstance(Container.class);
    container.setId(cID);
    NodeId nodeId = NodeId.newInstance(this.nmHost, this.nmPort);
    container.setNodeId(nodeId);
    container.setContainerToken(null);
    container.setNodeHttpAddress(this.nmHost + ":" + this.nmHttpPort);
    // send the container-assigned event to task attempt

    if (event.getAttemptID().getTaskId().getTaskType() == TaskType.MAP) {
      JobCounterUpdateEvent jce =
          new JobCounterUpdateEvent(event.getAttemptID().getTaskId()
              .getJobId());
      // TODO Setting OTHER_LOCAL_MAP for now.
      jce.addCounterUpdate(JobCounter.OTHER_LOCAL_MAPS, 1);
      eventHandler.handle(jce);
    }
    eventHandler.handle(new TaskAttemptContainerAssignedEvent(
        event.getAttemptID(), container, applicationACLs));
  }
}

源代码25 项目： big-c 文件： TaskAttemptImpl.java

private static void updateMillisCounters(JobCounterUpdateEvent jce,
    TaskAttemptImpl taskAttempt) {
  TaskType taskType = taskAttempt.getID().getTaskId().getTaskType();
  long duration = (taskAttempt.getFinishTime() - taskAttempt.getLaunchTime());
  int mbRequired =
      taskAttempt.getMemoryRequired(taskAttempt.conf, taskType);
  int vcoresRequired = taskAttempt.getCpuRequired(taskAttempt.conf, taskType);

  int minSlotMemSize = taskAttempt.conf.getInt(
    YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB,
    YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB);

  int simSlotsRequired =
      minSlotMemSize == 0 ? 0 : (int) Math.ceil((float) mbRequired
          / minSlotMemSize);

  if (taskType == TaskType.MAP) {
    jce.addCounterUpdate(JobCounter.SLOTS_MILLIS_MAPS, simSlotsRequired * duration);
    jce.addCounterUpdate(JobCounter.MB_MILLIS_MAPS, duration * mbRequired);
    jce.addCounterUpdate(JobCounter.VCORES_MILLIS_MAPS, duration * vcoresRequired);
    jce.addCounterUpdate(JobCounter.MILLIS_MAPS, duration);
  } else {
    jce.addCounterUpdate(JobCounter.SLOTS_MILLIS_REDUCES, simSlotsRequired * duration);
    jce.addCounterUpdate(JobCounter.MB_MILLIS_REDUCES, duration * mbRequired);
    jce.addCounterUpdate(JobCounter.VCORES_MILLIS_REDUCES, duration * vcoresRequired);
    jce.addCounterUpdate(JobCounter.MILLIS_REDUCES, duration);
  }
}

源代码26 项目： big-c 文件： TaskAttemptImpl.java

private static JobCounterUpdateEvent createJobCounterUpdateEventTAFailed(
    TaskAttemptImpl taskAttempt, boolean taskAlreadyCompleted) {
  TaskType taskType = taskAttempt.getID().getTaskId().getTaskType();
  JobCounterUpdateEvent jce = new JobCounterUpdateEvent(taskAttempt.getID().getTaskId().getJobId());
  
  if (taskType == TaskType.MAP) {
    jce.addCounterUpdate(JobCounter.NUM_FAILED_MAPS, 1);
  } else {
    jce.addCounterUpdate(JobCounter.NUM_FAILED_REDUCES, 1);
  }
  if (!taskAlreadyCompleted) {
    updateMillisCounters(jce, taskAttempt);
  }
  return jce;
}

源代码27 项目： big-c 文件： TaskAttemptImpl.java

private static JobCounterUpdateEvent createJobCounterUpdateEventTAKilled(
    TaskAttemptImpl taskAttempt, boolean taskAlreadyCompleted) {
  TaskType taskType = taskAttempt.getID().getTaskId().getTaskType();
  JobCounterUpdateEvent jce = new JobCounterUpdateEvent(taskAttempt.getID().getTaskId().getJobId());
  
  if (taskType == TaskType.MAP) {
    jce.addCounterUpdate(JobCounter.NUM_KILLED_MAPS, 1);
  } else {
    jce.addCounterUpdate(JobCounter.NUM_KILLED_REDUCES, 1);
  }
  if (!taskAlreadyCompleted) {
    updateMillisCounters(jce, taskAttempt);
  }
  return jce;
}

源代码28 项目： big-c 文件： JobHistoryEventHandler.java

private void setSummarySlotSeconds(JobSummary summary, Counters allCounters) {

    Counter slotMillisMapCounter = allCounters
      .findCounter(JobCounter.SLOTS_MILLIS_MAPS);
    if (slotMillisMapCounter != null) {
      summary.setMapSlotSeconds(slotMillisMapCounter.getValue() / 1000);
    }

    Counter slotMillisReduceCounter = allCounters
      .findCounter(JobCounter.SLOTS_MILLIS_REDUCES);
    if (slotMillisReduceCounter != null) {
      summary.setReduceSlotSeconds(slotMillisReduceCounter.getValue() / 1000);
    }
  }

源代码29 项目： big-c 文件： TestMRJobs.java

protected void verifyRandomWriterCounters(Job job)
    throws InterruptedException, IOException {
  Counters counters = job.getCounters();
  Assert.assertEquals(3, counters.findCounter(JobCounter.OTHER_LOCAL_MAPS)
      .getValue());
  Assert.assertEquals(3, counters.findCounter(JobCounter.TOTAL_LAUNCHED_MAPS)
      .getValue());
  Assert
      .assertTrue(counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS) != null
          && counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS).getValue() != 0);
}

源代码30 项目： big-c 文件： TestMRJobs.java

protected void verifyFailingMapperCounters(Job job)
    throws InterruptedException, IOException {
  Counters counters = job.getCounters();
  Assert.assertEquals(2, counters.findCounter(JobCounter.OTHER_LOCAL_MAPS)
      .getValue());
  Assert.assertEquals(2, counters.findCounter(JobCounter.TOTAL_LAUNCHED_MAPS)
      .getValue());
  Assert.assertEquals(2, counters.findCounter(JobCounter.NUM_FAILED_MAPS)
      .getValue());
  Assert
      .assertTrue(counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS) != null
          && counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS).getValue() != 0);
}

类所在包

org.apache.hadoop.mapreduce

类方法

NUM_FAILED_MAPS

同包方法

org.apache.hadoop.mapreduce.Job

org.apache.hadoop.mapreduce.lib.input.FileInputFormat

org.apache.hadoop.mapreduce.lib.output.FileOutputFormat

org.apache.hadoop.mapreduce.Mapper

org.apache.hadoop.mapreduce.TaskAttemptContext

org.apache.hadoop.mapreduce.InputSplit

org.apache.hadoop.mapreduce.JobContext