org.apache.hadoop.mapreduce.JobID#forName ( )源码实例Demo

HistoryEvent maybeEmitEvent(ParsedLine line, String jobIDName,
    HistoryEventEmitter thatg) {
  JobID jobID = JobID.forName(jobIDName);

  if (jobIDName == null) {
    return null;
  }

  String priority = line.get("JOB_PRIORITY");

  if (priority != null) {
    return new JobPriorityChangeEvent(jobID, JobPriority.valueOf(priority));
  }

  return null;
}
 

HistoryEvent maybeEmitEvent(ParsedLine line, String jobIDName,
    HistoryEventEmitter thatg) {
  if (jobIDName == null) {
    return null;
  }

  JobID jobID = JobID.forName(jobIDName);

  String launchTime = line.get("LAUNCH_TIME");

  if (launchTime != null) {
    Job20LineHistoryEventEmitter that =
        (Job20LineHistoryEventEmitter) thatg;
    return new JobInfoChangeEvent(jobID, that.originalSubmitTime, Long
        .parseLong(launchTime));
  }

  return null;
}
 

@Test
public void testJobNamePercentEncoding() throws IOException {
  JobIndexInfo info = new JobIndexInfo();
  JobID oldJobId = JobID.forName(JOB_ID);
  JobId jobId = TypeConverter.toYarn(oldJobId);
  info.setJobId(jobId);
  info.setSubmitTime(Long.parseLong(SUBMIT_TIME));
  info.setUser(USER_NAME);
  info.setJobName(JOB_NAME_WITH_DELIMITER);
  info.setFinishTime(Long.parseLong(FINISH_TIME));
  info.setNumMaps(Integer.parseInt(NUM_MAPS));
  info.setNumReduces(Integer.parseInt(NUM_REDUCES));
  info.setJobStatus(JOB_STATUS);
  info.setQueueName(QUEUE_NAME);
  info.setJobStartTime(Long.parseLong(JOB_START_TIME));

  String jobHistoryFile = FileNameIndexUtils.getDoneFileName(info);
  Assert.assertTrue("Job name not encoded correctly into job history file",
      jobHistoryFile.contains(JOB_NAME_WITH_DELIMITER_ESCAPE));
}
 

HistoryEvent maybeEmitEvent(ParsedLine line, String jobIDName,
    HistoryEventEmitter thatg) {
  if (jobIDName == null) {
    return null;
  }

  JobID jobID = JobID.forName(jobIDName);

  String launchTime = line.get("LAUNCH_TIME");
  String status = line.get("JOB_STATUS");
  String totalMaps = line.get("TOTAL_MAPS");
  String totalReduces = line.get("TOTAL_REDUCES");
  String uberized = line.get("UBERIZED");

  if (launchTime != null && totalMaps != null && totalReduces != null) {
    return new JobInitedEvent(jobID, Long.parseLong(launchTime), Integer
        .parseInt(totalMaps), Integer.parseInt(totalReduces), status,
        Boolean.parseBoolean(uberized));
  }

  return null;
}
 

@Test
public void testJobNamePercentEncoding() throws IOException {
  JobIndexInfo info = new JobIndexInfo();
  JobID oldJobId = JobID.forName(JOB_ID);
  JobId jobId = TypeConverter.toYarn(oldJobId);
  info.setJobId(jobId);
  info.setSubmitTime(Long.parseLong(SUBMIT_TIME));
  info.setUser(USER_NAME);
  info.setJobName(JOB_NAME_WITH_DELIMITER);
  info.setFinishTime(Long.parseLong(FINISH_TIME));
  info.setNumMaps(Integer.parseInt(NUM_MAPS));
  info.setNumReduces(Integer.parseInt(NUM_REDUCES));
  info.setJobStatus(JOB_STATUS);
  info.setQueueName(QUEUE_NAME);
  info.setJobStartTime(Long.parseLong(JOB_START_TIME));

  String jobHistoryFile = FileNameIndexUtils.getDoneFileName(info);
  Assert.assertTrue("Job name not encoded correctly into job history file",
      jobHistoryFile.contains(JOB_NAME_WITH_DELIMITER_ESCAPE));
}
 

@Override
public void cancel(String jobId) throws IOException {
  JobID id = JobID.forName(jobId);
  Cluster cluster = new Cluster(this.getConf());
  try {
    Job job = cluster.getJob(id);
    if (job == null) {
      LOG.error("No job found for " + id);
      // should we throw exception
      return;
    }
    if (job.isComplete() || job.isRetired()) {
      return;
    }

    job.killJob();
    LOG.debug("Killed copy job " + id);
  } catch (InterruptedException e) {
    throw new IOException(e);
  }
}
 

@Test
public void testQueueNamePercentEncoding() throws IOException {
  JobIndexInfo info = new JobIndexInfo();
  JobID oldJobId = JobID.forName(JOB_ID);
  JobId jobId = TypeConverter.toYarn(oldJobId);
  info.setJobId(jobId);
  info.setSubmitTime(Long.parseLong(SUBMIT_TIME));
  info.setUser(USER_NAME);
  info.setJobName(JOB_NAME);
  info.setFinishTime(Long.parseLong(FINISH_TIME));
  info.setNumMaps(Integer.parseInt(NUM_MAPS));
  info.setNumReduces(Integer.parseInt(NUM_REDUCES));
  info.setJobStatus(JOB_STATUS);
  info.setQueueName(QUEUE_NAME_WITH_DELIMITER);
  info.setJobStartTime(Long.parseLong(JOB_START_TIME));

  String jobHistoryFile = FileNameIndexUtils.getDoneFileName(info);
  Assert.assertTrue("Queue name not encoded correctly into job history file",
      jobHistoryFile.contains(QUEUE_NAME_WITH_DELIMITER_ESCAPE));
}
 

@Override
protected Job createJob(Configuration conf, JobStateInternal forcedState,
    String diagnostic) {
  JobImpl jobImpl = mock(JobImpl.class);
  when(jobImpl.getInternalState()).thenReturn(this.jobStateInternal);
  when(jobImpl.getAllCounters()).thenReturn(new Counters());
  JobID jobID = JobID.forName("job_1234567890000_0001");
  JobId jobId = TypeConverter.toYarn(jobID);
  when(jobImpl.getID()).thenReturn(jobId);
  ((AppContext) getContext())
      .getAllJobs().put(jobImpl.getID(), jobImpl);
  return jobImpl;
}
 

@Test
public void testListAttemptIdsWithInvalidInputs() throws Exception {
  JobID jobId = JobID.forName(jobIdStr);
  Cluster mockCluster = mock(Cluster.class);
  Job job = mock(Job.class);
  CLI cli = spy(new CLI());

  doReturn(mockCluster).when(cli).createCluster();
  when(mockCluster.getJob(jobId)).thenReturn(job);

  int retCode_JOB_SETUP = cli.run(new String[] { "-list-attempt-ids",
      jobIdStr, "JOB_SETUP", "running" });

  int retCode_JOB_CLEANUP = cli.run(new String[] { "-list-attempt-ids",
      jobIdStr, "JOB_CLEANUP", "running" });

  int retCode_invalidTaskState = cli.run(new String[] { "-list-attempt-ids",
      jobIdStr, "REDUCE", "complete" });

  assertEquals("JOB_SETUP is an invalid input,exit code should be -1", -1,
      retCode_JOB_SETUP);
  assertEquals("JOB_CLEANUP is an invalid input,exit code should be -1", -1,
      retCode_JOB_CLEANUP);
  assertEquals("complete is an invalid input,exit code should be -1", -1,
      retCode_invalidTaskState);

}
 

@Override
protected Job createJob(Configuration conf, JobStateInternal forcedState,
    String diagnostic) {
  JobImpl jobImpl = mock(JobImpl.class);
  when(jobImpl.getInternalState()).thenReturn(this.jobStateInternal);
  when(jobImpl.getAllCounters()).thenReturn(new Counters());
  JobID jobID = JobID.forName("job_1234567890000_0001");
  JobId jobId = TypeConverter.toYarn(jobID);
  when(jobImpl.getID()).thenReturn(jobId);
  ((AppContext) getContext())
      .getAllJobs().put(jobImpl.getID(), jobImpl);
  return jobImpl;
}
 

@Test
public void testJobHistoryFileNameBackwardsCompatible() throws IOException {
  JobID oldJobId = JobID.forName(JOB_ID);
  JobId jobId = TypeConverter.toYarn(oldJobId);

  long submitTime = Long.parseLong(SUBMIT_TIME);
  long finishTime = Long.parseLong(FINISH_TIME);
  int numMaps = Integer.parseInt(NUM_MAPS);
  int numReduces = Integer.parseInt(NUM_REDUCES);

  String jobHistoryFile = String.format(OLD_JOB_HISTORY_FILE_FORMATTER,
      JOB_ID,
      SUBMIT_TIME,
      USER_NAME,
      JOB_NAME,
      FINISH_TIME,
      NUM_MAPS,
      NUM_REDUCES,
      JOB_STATUS);

  JobIndexInfo info = FileNameIndexUtils.getIndexInfo(jobHistoryFile);
  Assert.assertEquals("Job id incorrect after decoding old history file",
      jobId, info.getJobId());
  Assert.assertEquals("Submit time incorrect after decoding old history file",
      submitTime, info.getSubmitTime());
  Assert.assertEquals("User incorrect after decoding old history file",
      USER_NAME, info.getUser());
  Assert.assertEquals("Job name incorrect after decoding old history file",
      JOB_NAME, info.getJobName());
  Assert.assertEquals("Finish time incorrect after decoding old history file",
      finishTime, info.getFinishTime());
  Assert.assertEquals("Num maps incorrect after decoding old history file",
      numMaps, info.getNumMaps());
  Assert.assertEquals("Num reduces incorrect after decoding old history file",
      numReduces, info.getNumReduces());
  Assert.assertEquals("Job status incorrect after decoding old history file",
      JOB_STATUS, info.getJobStatus());
  Assert.assertNull("Queue name incorrect after decoding old history file",
      info.getQueueName());
}
 

public void setDatum(Object oDatum) {
  this.datum = (JobFinished) oDatum;
  this.jobId = JobID.forName(datum.jobid.toString());
  this.finishTime = datum.finishTime;
  this.finishedMaps = datum.finishedMaps;
  this.finishedReduces = datum.finishedReduces;
  this.failedMaps = datum.failedMaps;
  this.failedReduces = datum.failedReduces;
  this.mapCounters = EventReader.fromAvro(datum.mapCounters);
  this.reduceCounters = EventReader.fromAvro(datum.reduceCounters);
  this.totalCounters = EventReader.fromAvro(datum.totalCounters);
}
 

@Test
public void testListAttemptIdsWithValidInput() throws Exception {
  JobID jobId = JobID.forName(jobIdStr);
  Cluster mockCluster = mock(Cluster.class);
  Job job = mock(Job.class);
  CLI cli = spy(new CLI());

  doReturn(mockCluster).when(cli).createCluster();
  when(job.getTaskReports(TaskType.MAP)).thenReturn(
      getTaskReports(jobId, TaskType.MAP));
  when(job.getTaskReports(TaskType.REDUCE)).thenReturn(
      getTaskReports(jobId, TaskType.REDUCE));
  when(mockCluster.getJob(jobId)).thenReturn(job);

  int retCode_MAP = cli.run(new String[] { "-list-attempt-ids", jobIdStr,
      "MAP", "running" });
  // testing case insensitive behavior
  int retCode_map = cli.run(new String[] { "-list-attempt-ids", jobIdStr,
      "map", "running" });

  int retCode_REDUCE = cli.run(new String[] { "-list-attempt-ids", jobIdStr,
      "REDUCE", "running" });

  int retCode_completed = cli.run(new String[] { "-list-attempt-ids",
      jobIdStr, "REDUCE", "completed" });

  assertEquals("MAP is a valid input,exit code should be 0", 0, retCode_MAP);
  assertEquals("map is a valid input,exit code should be 0", 0, retCode_map);
  assertEquals("REDUCE is a valid input,exit code should be 0", 0,
      retCode_REDUCE);
  assertEquals(
      "REDUCE and completed are a valid inputs to -list-attempt-ids,exit code should be 0",
      0, retCode_completed);

  verify(job, times(2)).getTaskReports(TaskType.MAP);
  verify(job, times(2)).getTaskReports(TaskType.REDUCE);
}
 

@Test
public void testCheckAccess() {
  // Create two unique users
  String user1 = System.getProperty("user.name");
  String user2 = user1 + "1234";
  UserGroupInformation ugi1 = UserGroupInformation.createRemoteUser(user1);
  UserGroupInformation ugi2 = UserGroupInformation.createRemoteUser(user2);

  // Create the job
  JobID jobID = JobID.forName("job_1234567890000_0001");
  JobId jobId = TypeConverter.toYarn(jobID);

  // Setup configuration access only to user1 (owner)
  Configuration conf1 = new Configuration();
  conf1.setBoolean(MRConfig.MR_ACLS_ENABLED, true);
  conf1.set(MRJobConfig.JOB_ACL_VIEW_JOB, "");

  // Verify access
  JobImpl job1 = new JobImpl(jobId, null, conf1, null, null, null, null, null,
      null, null, null, true, user1, 0, null, null, null, null);
  Assert.assertTrue(job1.checkAccess(ugi1, JobACL.VIEW_JOB));
  Assert.assertFalse(job1.checkAccess(ugi2, JobACL.VIEW_JOB));

  // Setup configuration access to the user1 (owner) and user2
  Configuration conf2 = new Configuration();
  conf2.setBoolean(MRConfig.MR_ACLS_ENABLED, true);
  conf2.set(MRJobConfig.JOB_ACL_VIEW_JOB, user2);

  // Verify access
  JobImpl job2 = new JobImpl(jobId, null, conf2, null, null, null, null, null,
      null, null, null, true, user1, 0, null, null, null, null);
  Assert.assertTrue(job2.checkAccess(ugi1, JobACL.VIEW_JOB));
  Assert.assertTrue(job2.checkAccess(ugi2, JobACL.VIEW_JOB));

  // Setup configuration access with security enabled and access to all
  Configuration conf3 = new Configuration();
  conf3.setBoolean(MRConfig.MR_ACLS_ENABLED, true);
  conf3.set(MRJobConfig.JOB_ACL_VIEW_JOB, "*");

  // Verify access
  JobImpl job3 = new JobImpl(jobId, null, conf3, null, null, null, null, null,
      null, null, null, true, user1, 0, null, null, null, null);
  Assert.assertTrue(job3.checkAccess(ugi1, JobACL.VIEW_JOB));
  Assert.assertTrue(job3.checkAccess(ugi2, JobACL.VIEW_JOB));

  // Setup configuration access without security enabled
  Configuration conf4 = new Configuration();
  conf4.setBoolean(MRConfig.MR_ACLS_ENABLED, false);
  conf4.set(MRJobConfig.JOB_ACL_VIEW_JOB, "");

  // Verify access
  JobImpl job4 = new JobImpl(jobId, null, conf4, null, null, null, null, null,
      null, null, null, true, user1, 0, null, null, null, null);
  Assert.assertTrue(job4.checkAccess(ugi1, JobACL.VIEW_JOB));
  Assert.assertTrue(job4.checkAccess(ugi2, JobACL.VIEW_JOB));

  // Setup configuration access without security enabled
  Configuration conf5 = new Configuration();
  conf5.setBoolean(MRConfig.MR_ACLS_ENABLED, true);
  conf5.set(MRJobConfig.JOB_ACL_VIEW_JOB, "");

  // Verify access
  JobImpl job5 = new JobImpl(jobId, null, conf5, null, null, null, null, null,
      null, null, null, true, user1, 0, null, null, null, null);
  Assert.assertTrue(job5.checkAccess(ugi1, null));
  Assert.assertTrue(job5.checkAccess(ugi2, null));
}
 

/** Get the job ID */
public JobID getJobId() { return JobID.forName(datum.jobid.toString()); }
 

/** Get the Job ID */
public JobID getJobId() { return JobID.forName(datum.jobid.toString()); }
 

/** Get the Job Id */
public JobID getJobId() { return JobID.forName(datum.jobid.toString()); }
 

/** Get the Job ID */
public JobID getJobId() { return JobID.forName(datum.jobid.toString()); }
 

/** Get the Job ID */
public JobID getJobId() { return JobID.forName(datum.jobid.toString()); }
 

/** Get the Job ID */
public JobID getJobId() { return JobID.forName(datum.jobid.toString()); }