Never called in MapRed jobs. Just for completion and test purposes
965196
@SuppressWarnings("rawtypes")
  @Override
	public int compare(ITuple w1, ITuple w2) {
		int schemaId1 = tupleMRConf.getSchemaIdByName(w1.getSchema().getName());
		int schemaId2 = tupleMRConf.getSchemaIdByName(w2.getSchema().getName());
		int[] indexes1 = serInfo.getGroupSchemaIndexTranslation(schemaId1);
		int[] indexes2 = serInfo.getGroupSchemaIndexTranslation(schemaId2);
	  Serializer[] serializers = serInfo.getGroupSchemaSerializers();
		return compare(w1.getSchema(), groupCriteria, w1, indexes1, w2, indexes2,serializers);
	}
890998
private Status runSparkJob(JavaSparkContext sc, PrintWriter reportWriter) {
    // Generate a list of integer for testing
    List<Integer> nums = IntStream.rangeClosed(1, mPartitions).boxed().collect(Collectors.toList());

    JavaRDD<Integer> dataSet = sc.parallelize(nums, mPartitions);

    // Run a Spark job to check whether Spark executors can recognize Alluxio
    JavaPairRDD<Status, String> extractedStatus = dataSet
        .mapToPair(s -> new Tuple2<>(CheckerUtils.performIntegrationChecks(),
            CheckerUtils.getLocalAddress()));

    // Merge the IP addresses that can/cannot recognize Alluxio
    JavaPairRDD<Status, String> mergeStatus = extractedStatus.reduceByKey((a, b)
        -> a.contains(b) ?  a : (b.contains(a) ? b : a + " " + b),
        (mPartitions < 10 ? 1 : mPartitions / 10));

    mSparkJobResult = mergeStatus.collect();

    Map<Status, List<String>> resultMap = new HashMap<>();
    for (Tuple2<Status, String> op : mSparkJobResult) {
      List<String> addresses = resultMap.getOrDefault(op._1, new ArrayList<>());
      addresses.add(op._2);
      resultMap.put(op._1, addresses);
    }

    return CheckerUtils.printNodesResults(resultMap, reportWriter);
  }
656402
function jobCallback(job, worker, index) {

        if (index < ALEXA_SIZE) {
            var line = lines[index].trim();

            var split = line.split(',');
            var id = parseInt(split[0]);
            var url = split[1];
            job({
                id : id,
                url : url
            }, function(err) {
                // Lets log if it worked
                total++;
                if (err) {
                    console.log('Problem  #' + worker.id + ': ' + err.message + ' for line: ' + line);
                } else {
                    console.log('    #' + worker.id + '  DONE: ' + line);
                    successful++;
                }
                if (total % 10 === 0) {
                    console.log('################## ' + successful + '/' + total + ' (success/crawled) ################################');
                }
            });
        } else {
            // no more content!
            job(null);
        }
    }
1050998
private static void fillInMissingMapOutputs(FileSystem fs, 
                                              TaskAttemptID taskId,
                                              int numMaps,
                                              JobConf conf) throws IOException {
    Class<? extends WritableComparable> keyClass
        = conf.getMapOutputKeyClass().asSubclass(WritableComparable.class);
    Class<? extends Writable> valueClass
        = conf.getMapOutputValueClass().asSubclass(Writable.class);
    MapOutputFile namer = new MapOutputFile(taskId.getJobID());
    namer.setConf(conf);
    for(int i=0; i<numMaps; i++) {
      Path f = namer.getInputFile(i, taskId);
      if (!fs.exists(f)) {
        LOG.info("Create missing input: " + f);
        SequenceFile.Writer out =
          SequenceFile.createWriter(fs, conf, f, keyClass, valueClass);
        out.close();
      }
    }    
  }
1283152
def _yarn_init(self, rm_address, requests_config, tags):
        
        running_apps = self._yarn_get_running_spark_apps(rm_address, requests_config, tags)

        # Report success after gathering all metrics from ResourceManager
        self.service_check(
            YARN_SERVICE_CHECK,
            AgentCheck.OK,
            tags=['url:%s' % rm_address] + tags,
            message='Connection to ResourceManager "%s" was successful' % rm_address,
        )

        return running_apps
254173
@Override
  public final void commitJob(org.apache.hadoop.mapreduce.JobContext context
                               ) throws IOException {
    commitJob((JobContext) context);
  }
812549
function (user, config, job, dirs, done) {
    var self = this;
    var tasks = [function (next) {
      var plugin = self.extensions.provider[job.project.provider.id];
      if (!plugin) {
        var msg = `Provider plugin '${job.project.provider.id}' not found in this environment!`;
        return next(new Error(msg));
      }
      var finished = t(function (err, provider) {
        if (provider) provider.id = job.project.provider.id;
        next(err, provider);
      });
      if (!plugin.hosted) {
        return plugin.init(dirs, job.providerConfig, job, finished);
      }
      plugin.init(dirs, user.account(job.project.provider).config, job.providerConfig, job, finished);
    }];
    // XXX: do we want/need more things in the context? See
    // extension-loader readme for what gets passed in to the
    // individual phase functions as "Phase context"
    var context = {
      baseDir: dirs.base,
      dataDir: dirs.data,
      cacheDir: dirs.cache
    };
    var anyPluginFailed = config.plugins.some(function (plugin) {
      if (false === plugin.enabled) {
        debug(`[runner:${self.id}] Found a disabled plugin: ${plugin.id} Project: ${job.project.name} Job ID: ${job._id}`);
        return;
      }
      if (!self.extensions.job[plugin.id]) {
        debug(`[runner:${self.id}] Error: Plugin not found ${plugin.id} Project: ${job.project.name} Job ID: ${job._id}`);
        debug('Plugin not found', plugin.id);
        done(new Error(`Plugin required but not found: ${plugin.id}`));
        return true;
      }
      debug(`Initializing plugin '${plugin.id}'...`);
      tasks.push(function (next) {
        var fn = self.extensions.job[plugin.id].init;
        var cb = t(function (err, obj) {
          if (obj) obj.id = plugin.id;
          next(err, obj);
        });
        switch (fn.length) {
        case 1:
          return fn(cb);
        case 2:
          return fn(plugin.config, cb);
        case 3:
          return fn(plugin.config, context, cb);
        case 4:
          return fn(plugin.config, job, context, cb);
        default:
          return fn(user.jobplugins && user.jobplugins[plugin.id], plugin.config, job, context, cb);
        }
      });
    });
    
    if (anyPluginFailed) return;
    
    async.parallel(tasks, function (err, results) {
      if (err) return done(err);
      done(null, {
        provider: results.shift(),
        jobplugins: results
      });
    });
  }
1221913
@SuppressWarnings("unchecked")
    private CompletionStage<List<V>> invokeMapBatchLoader(List<K> keys, BatchLoaderEnvironment environment) {
        CompletionStage<Map<K, V>> loadResult;
        Set<K> setOfKeys = new LinkedHashSet<>(keys);
        if (batchLoadFunction instanceof MappedBatchLoaderWithContext) {
            loadResult = ((MappedBatchLoaderWithContext<K, V>) batchLoadFunction).load(setOfKeys, environment);
        } else {
            loadResult = ((MappedBatchLoader<K, V>) batchLoadFunction).load(setOfKeys);
        }
        CompletionStage<Map<K, V>> mapBatchLoad = nonNull(loadResult, "Your batch loader function MUST return a non null CompletionStage promise");
        return mapBatchLoad.thenApply(map -> {
            List<V> values = new ArrayList<>();
            for (K key : keys) {
                V value = map.get(key);
                values.add(value);
            }
            return values;
        });
    }
143031
def absent(name,
           **kwargs):
    '''
    
    '''
    ret = {'name': name,
           'result': True,
           'changes': {},
           'comment': []}

    if __salt__['jenkins.job_exists'](name):
        try:
            __salt__['jenkins.delete_job'](name)
        except CommandExecutionError as exc:
            return _fail(ret, exc.strerror)
        else:
            ret['comment'] = 'Job \'{0}\' deleted.'.format(name)
    else:
        ret['comment'] = 'Job \'{0}\' already absent.'.format(name)
    return ret
1022981
def run_job():
    """"""
    async = get_current_async()
    async_options = async.get_options()

    job = async_options.get('job')
    if not job:
        raise Exception('This async contains no job to execute!')

    __, args, kwargs = job

    if args is None:
        args = ()

    if kwargs is None:
        kwargs = {}

    function = async._decorate_job()

    try:
        async.executing = True
        async.result = AsyncResult(payload=function(*args, **kwargs),
                                   status=AsyncResult.SUCCESS)
    except Abort as abort:
        logging.info('Async job was aborted: %r', abort)
        async.result = AsyncResult(status=AsyncResult.ABORT)

        # QUESTION: In this eventuality, we should probably tell the context we
        # are "complete" and let it handle completion checking.
        _handle_context_completion_check(async)
        return
    except AbortAndRestart as restart:
        logging.info('Async job was aborted and restarted: %r', restart)
        raise
    except BaseException as e:
        async.result = AsyncResult(payload=encode_exception(e),
                                   status=AsyncResult.ERROR)

    _handle_results(async_options)
    _handle_context_completion_check(async)
763742
def _callback(self, bjobid, result, grade, problems, tests, custom, archive, stdout, stderr):
        
        self._jobs_done[str(bjobid)] = (result, grade, problems, tests, custom, archive, stdout, stderr)
        self._waiting_jobs.remove(str(bjobid))
368642
def run_mapper(self, stdin=sys.stdin, stdout=sys.stdout):
        
        self.init_hadoop()
        self.init_mapper()
        outputs = self._map_input((line[:-1] for line in stdin))
        if self.reducer == NotImplemented:
            self.writer(outputs, stdout)
        else:
            self.internal_writer(outputs, stdout)
572130
public void doUpdates(MetricsContext unused) {
    // In case of running in LocalMode tracker == null
    if (tracker != null) {
      synchronized (tracker) {
        synchronized (this) {
          numRunningMaps = 0;
          numRunningReduces = 0;

          numWaitingMaps = 0;
          numWaitingReduces = 0;
          numTasksInMemory = 0;

          List<JobInProgress> jobs = tracker.getRunningJobs();
          for (JobInProgress jip : jobs) {
            for (TaskInProgress tip : jip.maps) {
              if (tip.isRunning()) {
                numRunningMaps++;
              } else if (tip.isRunnable()) {
                numWaitingMaps++;
              }
            }
            for (TaskInProgress tip : jip.reduces) {
              if (tip.isRunning()) {
                numRunningReduces++;
              } else if (tip.isRunnable()) {
                numWaitingReduces++;
              }

            }
            numTasksInMemory += jip.getTasks(TaskType.MAP).length;
            numTasksInMemory += jip.getTasks(TaskType.REDUCE).length;
          }

          // Get tracker metrics
          numTrackersDead = tracker.getDeadNodes().size();
          ClusterStatus cs = tracker.getClusterStatus(false);
          numTrackersExcluded = cs.getNumExcludedNodes();
        }
      }
    }
    synchronized (this) {
      metricsRecord.setMetric("map_slots", numMapSlots);
      metricsRecord.setMetric("reduce_slots", numReduceSlots);
      metricsRecord.incrMetric("blacklisted_maps", numBlackListedMapSlots);
      metricsRecord.incrMetric("blacklisted_reduces",
          numBlackListedReduceSlots);
      metricsRecord.incrMetric("jobs_submitted", numJobsSubmitted);
      metricsRecord.incrMetric("jobs_completed", numJobsCompleted);
      metricsRecord.setMetric("waiting_maps", numWaitingMaps);
      metricsRecord.setMetric("waiting_reduces", numWaitingReduces);

      metricsRecord.incrMetric("reserved_map_slots", numReservedMapSlots);
      metricsRecord.incrMetric("reserved_reduce_slots", numReservedReduceSlots);
      metricsRecord.incrMetric("occupied_map_slots", numOccupiedMapSlots);
      metricsRecord.incrMetric("occupied_reduce_slots", numOccupiedReduceSlots);

      metricsRecord.incrMetric("jobs_failed", numJobsFailed);
      metricsRecord.incrMetric("jobs_killed", numJobsKilled);

      metricsRecord.incrMetric("jobs_preparing", numJobsPreparing);
      metricsRecord.incrMetric("jobs_running", numJobsRunning);

      metricsRecord.setMetric("running_maps", numRunningMaps);
      metricsRecord.setMetric("running_reduces", numRunningReduces);

      metricsRecord.setMetric("num_tasks_in_memory", numTasksInMemory);

      metricsRecord.setMetric("trackers", numTrackers);
      metricsRecord.setMetric("trackers_blacklisted", numTrackersBlackListed);
      metricsRecord.setMetric("trackers_decommissioned",
          numTrackersDecommissioned);
      metricsRecord.setMetric("trackers_excluded", numTrackersExcluded);
      metricsRecord.setMetric("trackers_dead", numTrackersDead);

      metricsRecord.incrMetric("num_launched_jobs", numJobsLaunched);
      metricsRecord.incrMetric("total_submit_time", totalSubmitTime);

      aggregateJobStats.incrementMetricsAndReset(metricsRecord);

      // Update additional speculation stats for measuring the performance
      // of different kinds of speculation
      for (SpecStats.TaskType taskType : SpecStats.TaskType.values()) {
        for (SpecStats.SpecType specType : SpecStats.SpecType.values()) {
          for(SpecStats.StatType statType : SpecStats.StatType.values()) {
            String key = "speculation_by_" + specType.toString().toLowerCase()  
                + "_" + "rate_" + taskType.toString().toLowerCase() + "_" + 
                statType.toString().toLowerCase();
            long value = specStats.getStat(taskType, specType, statType);
            metricsRecord.setMetric(key, value);
          }
        }
      }

      
      for (Group group: countersToMetrics) {
        String groupName = group.getName();
        for (Counter counter : group) {
          String name = groupName + "_" + counter.getName();
          name = name.replaceAll("[^a-zA-Z_]", "_").toLowerCase();
          metricsRecord.incrMetric(name, counter.getValue());
        }
      }
      clearCounters();

      numJobsSubmitted = 0;
      numJobsCompleted = 0;
      numWaitingMaps = 0;
      numWaitingReduces = 0;
      numBlackListedMapSlots = 0;
      numBlackListedReduceSlots = 0;

      numReservedMapSlots = 0;
      numReservedReduceSlots = 0;
      numOccupiedMapSlots = 0;
      numOccupiedReduceSlots = 0;

      numJobsFailed = 0;
      numJobsKilled = 0;

      numJobsPreparing = 0;
      numJobsRunning = 0;

      numRunningMaps = 0;
      numRunningReduces = 0;

      numTrackers = 0;
      numTrackersBlackListed = 0;

      totalSubmitTime = 0;
      numJobsLaunched = 0;
    }
    metricsRecord.update();
  }
782484
def check_for_completion
      return unless all_work_units_complete?
      set_next_status
      outs = gather_outputs_from_work_units
      return queue_for_workers([outs]) if merging?
      if complete?
        update_attributes(:outputs => outs, :time => time_taken)
        CloudCrowd.log "Job ##{id} (#{action}) #{display_status}." unless ENV['RACK_ENV'] == 'test'
        CloudCrowd.defer { fire_callback } if callback_url
      end
      self
    end
51884
@Override
  JobDesc create(QualifiedJobId qualifiedJobId, long submitTimeMillis,
      Configuration jobConf) {
    // TODO: Get the actual values appropriate for the plain Hadoop jobs.

    String appId = getAppId(jobConf);

    long appSubmitTimeMillis = jobConf.getLong(Constants.MR_RUN_CONF_KEY,
        submitTimeMillis);


    return create(qualifiedJobId, jobConf, appId, Constants.UNKNOWN,
        Framework.NONE, appSubmitTimeMillis);
  }
534946
@FFDCIgnore(JobStoppingException.class)
    private void executeAndWaitForCompletion(PartitionPlanDescriptor currentPlan) throws JobRestartException {
        if (isStoppingStoppedOrFailed()) {
            logger.fine("Job already in "
                        + runtimeWorkUnitExecution.getWorkUnitJobContext().getBatchStatus().toString()
                        + " state, exiting from executeAndWaitForCompletion() before beginning execution");
            return;
        }

        List<Integer> partitionsToExecute = getPartitionNumbersToExecute(currentPlan);

        logger.fine("Partitions to execute in this run: " + partitionsToExecute
                    + ".  Total number of partitions in step: " + currentPlan.getNumPartitionsInPlan());

        List<Integer> startedPartitions = new ArrayList<Integer>();
        List<Integer> finishedPartitions = new ArrayList<Integer>();
        List<Throwable> analyzerExceptions = new ArrayList<Throwable>();

        // Keep looping until all partitions have finished.
        while (finishedPartitions.size() < partitionsToExecute.size()) {

            startPartitions(partitionsToExecute, startedPartitions, finishedPartitions, currentPlan);

            // Check that there are still un-finished partitions running.
            // If not, break out of the loop.
            if (finishedPartitions.size() >= partitionsToExecute.size()) {
                break;
            }

            //Break this loop when nextPartitionFinished is -1
            try {
                waitForNextPartitionToFinish(analyzerExceptions, finishedPartitions);
            } catch (JobStoppingException e) {
                // break the loop
                break;
            }
        }

        if (!analyzerExceptions.isEmpty()) {
            rollbackPartitionedStep();
            throw new BatchContainerRuntimeException("Exception previously thrown by Analyzer, rolling back step.", analyzerExceptions.get(0));
        }
    }
1320278
static String awaitComplete(CloudDNS api, Job job) {
    RetryableException retryableException = new RetryableException(
        format("Job %s did not complete. Check your logs.", job.id), null);
    Retryer retryer = new Retryer.Default(500, 1000, 30);

    while (true) {
      job = api.getStatus(job.id);

      if ("COMPLETED".equals(job.status)) {
        return job.resultId;
      } else if ("ERROR".equals(job.status)) {
        throw new IllegalStateException(
            format("Job %s failed with error: %s", job.id, job.errorDetails));
      }

      retryer.continueOrPropagate(retryableException);
    }
  }