code: private void prepareHadoopJob(List<WorkUnit> workUnits) throws IOException {
      TimingEvent mrJobSetupTimer = this.eventSubmitter.getTimingEvent(TimingEvent.RunJobTimings.MR_JOB_SETUP);
  
      // Add dependent jars/files
      addDependencies(this.job.getConfiguration());
  
      this.job.setJarByClass(MRJobLauncher.class);
      this.job.setMapperClass(TaskRunner.class);
  
      // The job is mapper-only
      this.job.setNumReduceTasks(0);
  
      this.job.setInputFormatClass(GobblinWorkUnitsInputFormat.class);
      this.job.setOutputFormatClass(GobblinOutputFormat.class);
      this.job.setMapOutputKeyClass(NullWritable.class);
      this.job.setMapOutputValueClass(NullWritable.class);
  
      // Set speculative execution
  
      this.job.setSpeculativeExecution(isSpeculativeExecutionEnabled(this.jobProps));
  
      this.job.getConfiguration().set("mapreduce.job.user.classpath.first", "true");
  
      // Job input path is where input work unit files are stored
  
      // Prepare job input
      prepareJobInput(workUnits);
      FileInputFormat.addInputPath(this.job, this.jobInputPath);
  
      // Job output path is where serialized task states are stored
      FileOutputFormat.setOutputPath(this.job, this.jobOutputPath);
  
      // Serialize source state to a file which will be picked up by the mappers
      serializeJobState(this.fs, this.mrJobDir, this.conf, this.jobContext.getJobState(), this.job);
  
      if (this.jobProps.containsKey(ConfigurationKeys.MR_JOB_MAX_MAPPERS_KEY)) {
        GobblinWorkUnitsInputFormat.setMaxMappers(this.job,
            Integer.parseInt(this.jobProps.getProperty(ConfigurationKeys.MR_JOB_MAX_MAPPERS_KEY)));
      }
  
      this.job.getConfiguration().set(GOBBLIN_JOB_INTERRUPT_PATH_KEY, this.interruptPath.toString());
  
      mrJobSetupTimer.stop();
    }
docs: Prepare the Hadoop MR job, including configuring the job and setting up the input/output paths.