Submit a job to the map/reduce cluster. All of the necessary modifications
to the job to run under pipes are made to the configuration.
@param conf the job to submit to the cluster (MODIFIED)
@throws IOException
553997
public static RunningJob runJob(JobConf conf) throws IOException {
    setupPipesJob(conf);
    return JobClient.runJob(conf);
  }
695343
@Override
  public void localizeConfiguration(JobConf conf) throws IOException {
    super.localizeConfiguration(conf);
    conf.setNumMapTasks(numMaps);
  }
900147
private void initialize(Configuration conf) throws IOException {
    InetSocketAddress socAddr = AvatarNode.getAddress(conf);
    int handlerCount = conf.getInt("hdfs.avatarnode.handler.count", 3);

    // create rpc server
    // no point to deserialize job file in Avatar node.
    this.server = RPC.getServer(this, socAddr.getAddress().getHostAddress(), socAddr.getPort(),
        handlerCount, false, conf, false);

    // The rpc-server port can be ephemeral... ensure we have the 
    // correct info
    this.serverAddress = this.server.getListenerAddress();
    LOG.info("AvatarNode up at: " + this.serverAddress);
    this.registerMBean();
    this.server.start();
  }
368335
public void submit(final Configuration runtimeConfiguration, final String jobName) throws Exception {
    final Configuration driverConfiguration = getDriverConfiguration(jobName);
    Tang.Factory.getTang().newInjector(runtimeConfiguration).getInstance(REEF.class).submit(driverConfiguration);
  }
192779
@Override
  public final 
  void setupTask(org.apache.hadoop.mapreduce.TaskAttemptContext taskContext
                 ) throws IOException {
    setupTask((TaskAttemptContext) taskContext);
  }
141552
protected synchronized void submit() {
    try {
      if (theJobConf.getBoolean("create.empty.dir.if.nonexist", false)) {
        FileSystem fs = FileSystem.get(theJobConf);
        Path inputPaths[] = FileInputFormat.getInputPaths(theJobConf);
        for (int i = 0; i < inputPaths.length; i++) {
          if (!fs.exists(inputPaths[i])) {
            try {
              fs.mkdirs(inputPaths[i]);
            } catch (IOException e) {

            }
          }
        }
      }
      RunningJob running = jc.submitJob(theJobConf);
      this.mapredJobID = running.getID();
      this.state = Job.RUNNING;
    } catch (IOException ioe) {
      this.state = Job.FAILED;
      this.message = StringUtils.stringifyException(ioe);
    }
  }
1201535
private void writeObject(ObjectOutputStream out) throws IOException {
		super.write(out);
		out.writeUTF(this.mapreduceInputFormat.getClass().getName());
		out.writeUTF(this.keyClass.getName());
		out.writeUTF(this.valueClass.getName());
		this.configuration.write(out);
	}
1340646
public static RunningJob runJob(JobConf job) throws IOException {
    JobClient jc = new JobClient(job);
    RunningJob rj = jc.submitJob(job);
    try {
      if (!jc.monitorAndPrintJob(job, rj)) {
        throw new IOException("Job failed!");
      }
    } catch (InterruptedException ie) {
      Thread.currentThread().interrupt();
    }
    return rj;
  }
189132
@SuppressWarnings("deprecation")
  private void addJars(Path jarFileDir, String jarFileList, Configuration conf) throws IOException {
    LocalFileSystem lfs = FileSystem.getLocal(conf);
    for (String jarFile : SPLITTER.split(jarFileList)) {
      Path srcJarFile = new Path(jarFile);
      FileStatus[] fileStatusList = lfs.globStatus(srcJarFile);

      for (FileStatus status : fileStatusList) {
        // For each FileStatus there are chances it could fail in copying at the first attempt, due to file-existence
        // or file-copy is ongoing by other job instance since all Gobblin jobs share the same jar file directory.
        // the retryCount is to avoid cases (if any) where retry is going too far and causes job hanging.
        int retryCount = 0;
        boolean shouldFileBeAddedIntoDC = true;
        Path destJarFile = calculateDestJarFile(status, jarFileDir);
        // Adding destJarFile into HDFS until it exists and the size of file on targetPath matches the one on local path.
        while (!this.fs.exists(destJarFile) || fs.getFileStatus(destJarFile).getLen() != status.getLen()) {
          try {
            if (this.fs.exists(destJarFile) && fs.getFileStatus(destJarFile).getLen() != status.getLen()) {
              Thread.sleep(WAITING_TIME_ON_IMCOMPLETE_UPLOAD);
              throw new IOException("Waiting for file to complete on uploading ... ");
            }
            // Set the first parameter as false for not deleting sourceFile
            // Set the second parameter as false for not overwriting existing file on the target, by default it is true.
            // If the file is preExisted but overwrite flag set to false, then an IOException if thrown.
            this.fs.copyFromLocalFile(false, false, status.getPath(), destJarFile);
          } catch (IOException | InterruptedException e) {
            LOG.warn("Path:" + destJarFile + " is not copied successfully. Will require retry.");
            retryCount += 1;
            if (retryCount >= this.jarFileMaximumRetry) {
              LOG.error("The jar file:" + destJarFile + "failed in being copied into hdfs", e);
              // If retry reaches upper limit, skip copying this file.
              shouldFileBeAddedIntoDC = false;
              break;
            }
          }
        }
        if (shouldFileBeAddedIntoDC) {
          // Then add the jar file on HDFS to the classpath
          LOG.info(String.format("Adding %s to classpath", destJarFile));
          DistributedCache.addFileToClassPath(destJarFile, conf, this.fs);
        }
      }
    }
  }
515918
public static void main(String[] argv) throws IOException {
    int ret;
    InstancedConfiguration conf = new InstancedConfiguration(ConfigurationUtils.defaults());

    if (!ConfigurationUtils.masterHostConfigured(conf) && argv.length > 0) {
      System.out.println(ConfigurationUtils
          .getMasterHostNotConfiguredMessage("Alluxio job shell"));
      System.exit(1);
    }

    try (JobShell shell = new JobShell(conf)) {
      ret = shell.run(argv);
    }
    System.exit(ret);
  }
714602
@SuppressWarnings("deprecation")
  private static JobConf createJobConfForCopyByChunk(Configuration conf) {
    JobConf jobconf = new JobConf(conf, DistCp.class);
    jobconf.setJobName(NAME);

    // turn off speculative execution, because DFS doesn't handle
    // multiple writers to the same file.
    jobconf.setReduceSpeculativeExecution(false);

    jobconf.setMapOutputKeyClass(Text.class);
    jobconf.setMapOutputValueClass(IntWritable.class);
    jobconf.setOutputKeyClass(Text.class);
    jobconf.setOutputValueClass(Text.class);

    jobconf.setInputFormat(CopyByChunkInputFormat.class);
    jobconf.setMapperClass(CopyFilesByChunkMapper.class);
    jobconf.setReducerClass(CopyFilesByChunkReducer.class);
    
    // Prevent the reducer from starting until all maps are done.
    jobconf.setInt("mapred.job.rushreduce.reduce.threshold", 0);
    jobconf.setFloat("mapred.reduce.slowstart.completed.maps", 1.0f);
    
    return jobconf;
  }
435592
private void writeObject(ObjectOutputStream out) throws IOException {
		super.write(out);
		out.writeUTF(this.mapreduceOutputFormat.getClass().getName());
		this.configuration.write(out);
	}
176385
public static Job createSubmittableJob(Configuration conf, String[] args)
  throws IOException {
    conf.setIfUnset("hbase.client.connection.impl",
      BigtableConfiguration.getConnectionClass().getName());
    conf.setIfUnset(BigtableOptionsFactory.BIGTABLE_RPC_TIMEOUT_MS_KEY, "60000");

    TableName tableName = TableName.valueOf(args[0]);
    conf.set(TABLE_NAME, tableName.getNameAsString());
    Path inputDir = new Path(args[1]);
    Job job = Job.getInstance(conf, conf.get(JOB_NAME_CONF_KEY, NAME + "_" + tableName));
    job.setJarByClass(Importer.class);
    FileInputFormat.setInputPaths(job, inputDir);
    // Randomize the splits to avoid hot spotting a single tablet server
    job.setInputFormatClass(ShuffledSequenceFileInputFormat.class);
    // Give the mappers enough work to do otherwise each split will be dominated by spinup time
    ShuffledSequenceFileInputFormat.setMinInputSplitSize(job, 1L*1024*1024*1024);
    String hfileOutPath = conf.get(BULK_OUTPUT_CONF_KEY);

    // make sure we get the filter in the jars
    try {
      Class<? extends Filter> filter = conf.getClass(FILTER_CLASS_CONF_KEY, null, Filter.class);
      if (filter != null) {
        TableMapReduceUtil.addDependencyJars(conf, filter);
      }
    } catch (Exception e) {
      throw new IOException(e);
    }

    if (hfileOutPath != null) {
      job.setMapperClass(KeyValueImporter.class);
      try (Connection conn = ConnectionFactory.createConnection(conf); 
          Table table = conn.getTable(tableName);
          RegionLocator regionLocator = conn.getRegionLocator(tableName)){
        job.setReducerClass(KeyValueSortReducer.class);
        Path outputDir = new Path(hfileOutPath);
        FileOutputFormat.setOutputPath(job, outputDir);
        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
        job.setMapOutputValueClass(KeyValue.class);
        HFileOutputFormat2.configureIncrementalLoad(job, table, regionLocator);
        TableMapReduceUtil.addDependencyJars(job.getConfiguration(),
            com.google.common.base.Preconditions.class);
      }
    } else {
      // No reducers.  Just write straight to table.  Call initTableReducerJob
      // because it sets up the TableOutputFormat.
      job.setMapperClass(Importer.class);
      //TableMapReduceUtil.initTableReducerJob(tableName.getNameAsString(), null, job);
      TableMapReduceUtil.initTableReducerJob(tableName.getNameAsString(), null, job, null, null, null, null, false);
      job.setNumReduceTasks(0);
    }
    return job;
  }
1112888
public static void setSerializations(Configuration conf) {
		String serializations = conf.get(HADOOP_IO_SERIALIZATIONS);

		LinkedList<String> list = new LinkedList<String>();

		if(serializations != null && !serializations.isEmpty())
			Collections.addAll(list, serializations.split(","));

		// required by MultiInputSplit
		String writable = WritableSerialization.class.getName();
		String tuple = TupleSerialization.class.getName();

		list.remove(writable);
		list.remove(tuple);

		list.addFirst(writable);
		list.addFirst(tuple);

		// make writable last
		conf.set(HADOOP_IO_SERIALIZATIONS, Util.join(list, ","));
	}
193334
public static void run(final Configuration conf) throws IOException {
		final String inputString = conf.get("input");
		if (inputString == null)
			throw new IllegalArgumentException(
				"String property \"input\" path not found in given Configuration");

		final FileSystem fs = FileSystem.get(conf);

		final Path input = new Path(inputString);

		SplittingBAMIndexer.index(
			fs.open(input),
			fs.create(input.suffix(OUTPUT_FILE_EXTENSION)),
			fs.getFileStatus(input).getLen(),
			conf.getInt("granularity", DEFAULT_GRANULARITY));
	}
390350
private void writeObject(ObjectOutputStream out) throws IOException {
		// serialize the parent fields and the final fields
		out.defaultWriteObject();

		// the job conf knows how to serialize itself
		jobConf.write(out);

		// write the input split
		hadoopInputSplit.write(out);
	}
47
@Override
	public void configure(Configuration parameters) {

		// enforce sequential configuration() calls
		synchronized (CONFIGURE_MUTEX) {
			if (mapreduceInputFormat instanceof Configurable) {
				((Configurable) mapreduceInputFormat).setConf(configuration);
			}
		}
	}