import org.apache.hadoop.streaming.StreamJob; //导入依赖的package包/类
public void addJob(int numReducers, boolean mapoutputCompressed,
    boolean outputCompressed, Size size, JobControl gridmix) {
  final String prop = String.format("streamSort.%sJobs.inputFiles", size);
  final String indir = getInputDirsFor(prop, size.defaultPath(VARINFLTEXT));
  final String outdir = addTSSuffix("perf-out/stream-out-dir-" + size);

  StringBuffer sb = new StringBuffer();
  sb.append("-input ").append(indir).append(" ");
  sb.append("-output ").append(outdir).append(" ");
  sb.append("-mapper cat ");
  sb.append("-reducer cat ");
  sb.append("-numReduceTasks ").append(numReducers);
  String[] args = sb.toString().split(" ");

  clearDir(outdir);
  try {
    JobConf jobconf = StreamJob.createJob(args);
    jobconf.setJobName("GridmixStreamingSorter." + size);
    jobconf.setCompressMapOutput(mapoutputCompressed);
    jobconf.setBoolean("mapred.output.compress", outputCompressed);
    Job job = new Job(jobconf);
    gridmix.addJob(job);
  } catch (Exception ex) {
    ex.printStackTrace();
  }
}
 

import org.apache.hadoop.streaming.StreamJob; //导入依赖的package包/类
private JobID getJobId(String [] runtimeArgs, String [] otherArgs) 
    throws IOException {
  JobID jobId = null;
  final RunStreamJob runSJ;
  StreamJob streamJob = new StreamJob();
  int counter = 0;
  JTClient jtClient = cluster.getJTClient();
  JobClient jobClient = jtClient.getClient();
  int totalJobs = jobClient.getAllJobs().length;
  String [] args = buildArgs(runtimeArgs, otherArgs);
  cleanup(outputDir, conf);
  conf.setBoolean("mapreduce.job.complete.cancel.delegation.tokens", false);
  runSJ = new RunStreamJob(conf, streamJob, args);
  runSJ.start();
  while (counter++ < 60) {
    if (jobClient.getAllJobs().length - totalJobs == 0) {
      UtilsForTests.waitFor(1000);
    } else if (jobClient.getAllJobs()[0].getRunState() == JobStatus.RUNNING) {
      jobId = jobClient.getAllJobs()[0].getJobID();
      break;
    } else {
      UtilsForTests.waitFor(1000);
    }
  }  
  return jobId;
}
 

import org.apache.hadoop.streaming.StreamJob; //导入依赖的package包/类
private JobID getJobIdOfRunningStreamJob(String [] runtimeArgs) 
    throws IOException {
  JobID jobId = null;
  StreamJob streamJob = new StreamJob();
  int counter = 0;
  jtClient = cluster.getJTClient();
  client = jtClient.getClient();
  int totalJobs = client.getAllJobs().length;
  String [] streamingArgs = generateArgs(runtimeArgs);
  cleanup(outputDir, conf);
  conf.setBoolean("mapreduce.job.complete.cancel.delegation.tokens", false);
  final RunStreamingJob streamJobThread = new RunStreamingJob(conf,
      streamJob,streamingArgs);
  streamJobThread.start();
  while (counter++ < 60) {
    if (client.getAllJobs().length - totalJobs == 0) {
      UtilsForTests.waitFor(1000);
    } else if (client.getAllJobs()[0].getRunState() == JobStatus.RUNNING) {
      jobId = client.getAllJobs()[0].getJobID();
      break;
    } else {
     UtilsForTests.waitFor(1000);
    }
  }  
  return jobId;
}
 

import org.apache.hadoop.streaming.StreamJob; //导入依赖的package包/类
public void addJob(int numReducers, boolean mapoutputCompressed,
    boolean outputCompressed, Size size, JobControl gridmix) {
  final String prop = String.format("streamSort.%sJobs.inputFiles", size);
  final String indir = 
    getInputDirsFor(prop, size.defaultPath(VARINFLTEXT));
  final String outdir = addTSSuffix("perf-out/stream-out-dir-" + size);

  StringBuffer sb = new StringBuffer();
  sb.append("-input ").append(indir).append(" ");
  sb.append("-output ").append(outdir).append(" ");
  sb.append("-mapper cat ");
  sb.append("-reducer cat ");
  sb.append("-numReduceTasks ").append(numReducers);
  String[] args = sb.toString().split(" ");

  clearDir(outdir);
  try {
    Configuration conf = StreamJob.createJob(args);
    conf.setBoolean(FileOutputFormat.COMPRESS, outputCompressed);
    conf.setBoolean(MRJobConfig.MAP_OUTPUT_COMPRESS, mapoutputCompressed);
    Job job = new Job(conf, "GridmixStreamingSorter." + size);
    ControlledJob cjob = new ControlledJob(job, null);
    gridmix.addJob(cjob);
  } catch (Exception ex) {
    ex.printStackTrace();
  }
}
 

import org.apache.hadoop.streaming.StreamJob; //导入依赖的package包/类
private void executeStreamingJob(boolean sameUser) throws Exception {
    conf = cluster.getConf();
  if (sameUser == true) {
    UserGroupInformation ugi = UserGroupInformation.getLoginUser();
    LOG.info("LoginUser:" + ugi);
  } else {
    conf.set("user.name","hadoop1");
    LOG.info("user name changed is :" + conf.get("user.name"));
  }

  if (conf.get("mapred.task.tracker.task-controller").
      equals("org.apache.hadoop.mapred.LinuxTaskController")) {
    StreamJob streamJob = new StreamJob();
    String shellFile = System.getProperty("user.dir") +
      "/src/test/system/scripts/StreamMapper.sh";
    String runtimeArgs [] = {
      "-D", "mapred.job.name=Streaming job",
      "-D", "mapred.map.tasks=1",
      "-D", "mapred.reduce.tasks=1",
      "-D", "mapred.tasktracker.tasks.sleeptime-before-sigkill=3000",
      "-input", inputDir.toString(),
      "-output", outputDir.toString(),
      "-mapper", "StreamMapper.sh",
      "-reducer","/bin/cat",
      "-file", shellFile
    };

    createInput(inputDir, conf);
    cleanup(outputDir, conf);

    //If job submtitted with same user, it should pass
    //If job submitted with different user, it should fail with assertion
    //the testcase should assert.
    if (sameUser == true) {
      Assert.assertEquals(0, ToolRunner.run(conf, streamJob, runtimeArgs));
    } else {
      if ((ToolRunner.run(conf, streamJob, runtimeArgs)) != 0) {
        LOG.info("Job failed as expected");
      } else {
        Assert.fail("Job passed with different user");
      }
    }
  }
}