Java 类org.apache.hadoop.mapred.JobClient.RawSplit 实例源码

/**
 * Constructor for MapTask
 */
public TaskInProgress(JobID jobid, String jobFile, 
                      RawSplit rawSplit, 
                      JobConf conf, 
                      JobInProgressTraits job, int partition,
                      int numSlotsRequired) {
  this.jobFile = jobFile;
  this.rawSplit = rawSplit;
  this.job = job;
  this.conf = conf;
  this.partition = partition;
  this.maxSkipRecords = SkipBadRecords.getMapperMaxSkipRecords(conf);
  this.numSlotsRequired = numSlotsRequired;
  setMaxTaskAttempts();
  init(jobid);
}

RawSplit[] getRawSplits(InputSplit[] splits) throws IOException {
  if (splits == null || splits.length != numMapTasks) {
    throw new IllegalArgumentException("Input split size mismatch: expected="
        + numMapTasks + ", actual=" + ((splits == null) ? -1 : splits.length));
  }

  RawSplit rawSplits[] = new RawSplit[splits.length];
  for (int i = 0; i < splits.length; i++) {
    try {
      rawSplits[i] = new RawSplit();
      rawSplits[i].setClassName(splits[i].getClass().getName());
      rawSplits[i].setDataLength(splits[i].getLength());
      rawSplits[i].setLocations(splits[i].getLocations());
    } catch (InterruptedException ie) {
      throw new IOException(ie);
    }
  }

  return rawSplits;
}

private int getClosestLocality(TaskTracker taskTracker, RawSplit split) {
  int locality = 2;

  Node taskTrackerNode = jobtracker
      .getNode(taskTracker.getStatus().getHost());
  if (taskTrackerNode == null) {
    throw new IllegalArgumentException(
        "Cannot determine network topology node for TaskTracker "
            + taskTracker.getTrackerName());
  }
  for (String location : split.getLocations()) {
    Node dataNode = jobtracker.getNode(location);
    if (dataNode == null) {
      throw new IllegalArgumentException(
          "Cannot determine network topology node for split location "
              + location);
    }
    locality = Math.min(locality, jobtracker.clusterMap.getDistance(
        taskTrackerNode, dataNode));
  }
  return locality;
}

/**
 * Constructor for MapTask
 */
public TaskInProgress(JobID jobid, String jobFile, 
                      RawSplit rawSplit, 
                      JobConf conf, 
                      JobInProgressTraits job, int partition,
                      int numSlotsRequired) {
  this.jobFile = jobFile;
  this.rawSplit = rawSplit;
  this.job = job;
  this.conf = conf;
  this.partition = partition;
  this.maxSkipRecords = SkipBadRecords.getMapperMaxSkipRecords(conf);
  this.numSlotsRequired = numSlotsRequired;
  setMaxTaskAttempts();
  init(jobid);
}

RawSplit[] getRawSplits(InputSplit[] splits) throws IOException {
  if (splits == null || splits.length != numMapTasks) {
    throw new IllegalArgumentException("Input split size mismatch: expected="
        + numMapTasks + ", actual=" + ((splits == null) ? -1 : splits.length));
  }

  RawSplit rawSplits[] = new RawSplit[splits.length];
  for (int i = 0; i < splits.length; i++) {
    try {
      rawSplits[i] = new RawSplit();
      rawSplits[i].setClassName(splits[i].getClass().getName());
      rawSplits[i].setDataLength(splits[i].getLength());
      rawSplits[i].setLocations(splits[i].getLocations());
    } catch (InterruptedException ie) {
      throw new IOException(ie);
    }
  }

  return rawSplits;
}

private int getClosestLocality(TaskTracker taskTracker, RawSplit split) {
  int locality = 2;

  Node taskTrackerNode = jobtracker
      .getNode(taskTracker.getStatus().getHost());
  if (taskTrackerNode == null) {
    throw new IllegalArgumentException(
        "Cannot determine network topology node for TaskTracker "
            + taskTracker.getTrackerName());
  }
  for (String location : split.getLocations()) {
    Node dataNode = jobtracker.getNode(location);
    if (dataNode == null) {
      throw new IllegalArgumentException(
          "Cannot determine network topology node for split location "
              + location);
    }
    locality = Math.min(locality, jobtracker.clusterMap.getDistance(
        taskTrackerNode, dataNode));
  }
  return locality;
}

public void testResourceEstimator() throws Exception {
  final int maps = 100;
  final int reduces = 2;
  final int singleMapOutputSize = 1000;
  JobConf jc = new JobConf();
  JobID jid = new JobID("testJT", 0);
  jc.setNumMapTasks(maps);
  jc.setNumReduceTasks(reduces);

  JobInProgress jip = new JobInProgress(jid, jc, 
      UtilsForTests.getJobTracker());
  //unfortunately, we can't set job input size from here.
  ResourceEstimator re = new ResourceEstimator(jip);

  for(int i = 0; i < maps / 10 ; ++i) {

    long estOutSize = re.getEstimatedMapOutputSize();
    System.out.println(estOutSize);
    assertEquals(0, estOutSize);

    TaskStatus ts = new MapTaskStatus();
    ts.setOutputSize(singleMapOutputSize);
    RawSplit split = new RawSplit();
    split.setDataLength(0);
    TaskInProgress tip = 
      new TaskInProgress(jid, "", split, jc, jip, 0, 1);
    re.updateWithCompletedTask(ts, tip);
  }
  assertEquals(2* singleMapOutputSize, re.getEstimatedMapOutputSize());
  assertEquals(2* singleMapOutputSize * maps / reduces, re.getEstimatedReduceInputSize());

}

@Override
public synchronized void initTasks() throws IOException {

  RawSplit[] splits = createSplits();
  numMapTasks = splits.length;
  createMapTasks(splits);
  nonRunningMapCache = createCache(splits, maxLevel);
  createReduceTasks();
  tasksInited.set(true);
  this.status.setRunState(JobStatus.RUNNING);
}

RawSplit[] createSplits(){
  RawSplit[] splits = new RawSplit[numMapTasks];
  for (int i = 0; i < numMapTasks; i++) {
    splits[i] = new RawSplit();
    splits[i].setLocations(new String[0]);
  }
  return splits;
}

protected void createMapTasks(RawSplit[] splits) {
  maps = new TaskInProgress[numMapTasks];
  for (int i = 0; i < numMapTasks; i++) {
    maps[i] = new TaskInProgress(getJobID(), "test", 
        splits[i], getJobConf(), this, i, 1);
  }
}

public void testResourceEstimator() throws Exception {
  final int maps = 100;
  final int reduces = 2;
  final int singleMapOutputSize = 1000;
  JobConf jc = new JobConf();
  JobID jid = new JobID("testJT", 0);
  jc.setNumMapTasks(maps);
  jc.setNumReduceTasks(reduces);

  JobInProgress jip = new JobInProgress(jid, jc, 
      UtilsForTests.getJobTracker());
  //unfortunately, we can't set job input size from here.
  ResourceEstimator re = new ResourceEstimator(jip);

  for(int i = 0; i < maps / 10 ; ++i) {

    long estOutSize = re.getEstimatedMapOutputSize();
    System.out.println(estOutSize);
    assertEquals(0, estOutSize);

    TaskStatus ts = new MapTaskStatus();
    ts.setOutputSize(singleMapOutputSize);
    RawSplit split = new RawSplit();
    split.setDataLength(0);
    TaskInProgress tip = 
      new TaskInProgress(jid, "", split, jc, jip, 0, 1);
    re.updateWithCompletedTask(ts, tip);
  }
  assertEquals(2* singleMapOutputSize, re.getEstimatedMapOutputSize());
  assertEquals(2* singleMapOutputSize * maps / reduces, re.getEstimatedReduceInputSize());

}

@Override
public synchronized void initTasks() throws IOException {

  RawSplit[] splits = createSplits();
  numMapTasks = splits.length;
  createMapTasks(splits);
  nonRunningMapCache = createCache(splits, maxLevel);
  createReduceTasks();
  tasksInited.set(true);
  this.status.setRunState(JobStatus.RUNNING);
}

RawSplit[] createSplits(){
  RawSplit[] splits = new RawSplit[numMapTasks];
  for (int i = 0; i < numMapTasks; i++) {
    splits[i] = new RawSplit();
    splits[i].setLocations(new String[0]);
  }
  return splits;
}

protected void createMapTasks(RawSplit[] splits) {
  maps = new TaskInProgress[numMapTasks];
  for (int i = 0; i < numMapTasks; i++) {
    maps[i] = new TaskInProgress(getJobID(), "test", 
        splits[i], getJobConf(), this, i, 1);
  }
}

/**
 * Constructor for MapTask
 */
public TaskInProgress(JobID jobid, String jobFile, 
                      RawSplit rawSplit, 
                      JobTracker jobtracker, JobConf conf, 
                      JobInProgress job, int partition) {
  this.jobFile = jobFile;
  this.rawSplit = rawSplit;
  this.jobtracker = jobtracker;
  this.job = job;
  this.conf = conf;
  this.partition = partition;
  this.maxSkipRecords = SkipBadRecords.getMapperMaxSkipRecords(conf);
  setMaxTaskAttempts();
  init(jobid);
}

public void testResourceEstimator() throws Exception {
  final int maps = 100;
  final int reduces = 2;
  final int singleMapOutputSize = 1000;
  JobConf jc = new JobConf();
  JobID jid = new JobID("testJT", 0);
  jc.setNumMapTasks(maps);
  jc.setNumReduceTasks(reduces);

  JobInProgress jip = new JobInProgress(jid, jc);
  //unfortunately, we can't set job input size from here.
  ResourceEstimator re = new ResourceEstimator(jip);

  for(int i = 0; i < maps / 10 ; ++i) {

    long estOutSize = re.getEstimatedMapOutputSize();
    System.out.println(estOutSize);
    assertEquals(0, estOutSize);

    TaskStatus ts = new MapTaskStatus();
    ts.setOutputSize(singleMapOutputSize);
    RawSplit split = new RawSplit();
    split.setDataLength(0);
    TaskInProgress tip = new TaskInProgress(jid, "", split, null, jc, jip, 0);
    re.updateWithCompletedTask(ts, tip);
  }
  assertEquals(2* singleMapOutputSize, re.getEstimatedMapOutputSize());
  assertEquals(2* singleMapOutputSize * maps / reduces, re.getEstimatedReduceInputSize());

}

/**
 * Constructor for MapTask
 */
public TaskInProgress(JobID jobid, String jobFile, 
                      RawSplit rawSplit, 
                      JobTracker jobtracker, JobConf conf, 
                      JobInProgress job, int partition) {
  this.jobFile = jobFile;
  this.rawSplit = rawSplit;
  this.jobtracker = jobtracker;
  this.job = job;
  this.conf = conf;
  this.partition = partition;
  this.maxSkipRecords = SkipBadRecords.getMapperMaxSkipRecords(conf);
  setMaxTaskAttempts();
  init(jobid);
}

public void testResourceEstimator() throws Exception {
  final int maps = 100;
  final int reduces = 2;
  final int singleMapOutputSize = 1000;
  JobConf jc = new JobConf();
  JobID jid = new JobID("testJT", 0);
  jc.setNumMapTasks(maps);
  jc.setNumReduceTasks(reduces);

  JobInProgress jip = new JobInProgress(jid, jc);
  //unfortunately, we can't set job input size from here.
  ResourceEstimator re = new ResourceEstimator(jip);

  for(int i = 0; i < maps / 10 ; ++i) {

    long estOutSize = re.getEstimatedMapOutputSize();
    System.out.println(estOutSize);
    assertEquals(0, estOutSize);

    TaskStatus ts = new MapTaskStatus();
    ts.setOutputSize(singleMapOutputSize);
    RawSplit split = new RawSplit();
    split.setDataLength(0);
    TaskInProgress tip = new TaskInProgress(jid, "", split, null, jc, jip, 0);
    re.updateWithCompletedTask(ts, tip);
  }
  assertEquals(2* singleMapOutputSize, re.getEstimatedMapOutputSize());
  assertEquals(2* singleMapOutputSize * maps / reduces, re.getEstimatedReduceInputSize());

}

public FakeTaskInProgress(JobID jobId, String jobFile, RawSplit emptySplit,
    JobTracker jobTracker, JobConf jobConf,
    JobInProgress job, int partition, int numSlotsRequired) {
  super(jobId, jobFile, emptySplit, jobConf, job,
        partition, numSlotsRequired);
}

public FakeTaskInProgress(JobID jobId, String jobFile, RawSplit emptySplit,
    JobTracker jobTracker, JobConf jobConf,
    JobInProgress job, int partition, int numSlotsRequired) {
  super(jobId, jobFile, emptySplit, jobConf, job,
        partition, numSlotsRequired);
}