We write settings to ~/.fscrawler/{job_name}/_status.json
@param jobname is the job_name
@param job Status file to write
@throws IOException in case of error while reading
505258
public void write(String jobname, FsJob job) throws IOException {
        writeFile(jobname, FILENAME, FsJobParser.toJson(job));
    }
1168118
public void configure(JobConf job) {
    this.inputFile = job.get("map.input.file");
    maxNumItems = job.getLong("aggregate.max.num.unique.values",
                              Long.MAX_VALUE);
  }
1138240
public void setStoreLocation(String location, Job job) throws IOException
    {
        conf = HadoopCompat.getConfiguration(job);
        setLocationFromUri(location);

        if (username != null && password != null)
            ConfigHelper.setOutputKeyspaceUserNameAndPassword(conf, username, password);
        if (splitSize > 0)
            ConfigHelper.setInputSplitSize(conf, splitSize);
        if (partitionerClass!= null)
            ConfigHelper.setOutputPartitioner(conf, partitionerClass);
        if (rpcPort != null)
        {
            ConfigHelper.setOutputRpcPort(conf, rpcPort);
            ConfigHelper.setInputRpcPort(conf, rpcPort);
        }
        if (initHostAddress != null)
        {
            ConfigHelper.setOutputInitialAddress(conf, initHostAddress);
            ConfigHelper.setInputInitialAddress(conf, initHostAddress);
        }

        ConfigHelper.setOutputColumnFamily(conf, keyspace, column_family);
        CqlConfigHelper.setOutputCql(conf, outputQuery);

        setConnectionInformation();

        if (ConfigHelper.getOutputRpcPort(conf) == 0)
            throw new IOException("PIG_OUTPUT_RPC_PORT or PIG_RPC_PORT environment variable not set");
        if (ConfigHelper.getOutputInitialAddress(conf) == null)
            throw new IOException("PIG_OUTPUT_INITIAL_ADDRESS or PIG_INITIAL_ADDRESS environment variable not set");
        if (ConfigHelper.getOutputPartitioner(conf) == null)
            throw new IOException("PIG_OUTPUT_PARTITIONER or PIG_PARTITIONER environment variable not set");

        initSchema(storeSignature);
    }
49279
public void setJobSummaryList(java.util.Collection<JobSummary> jobSummaryList) {
        if (jobSummaryList == null) {
            this.jobSummaryList = null;
            return;
        }

        this.jobSummaryList = new java.util.ArrayList<JobSummary>(jobSummaryList);
    }
729049
private void updateFsJob(String jobName, LocalDateTime scanDate) throws Exception {
        // We need to round that latest date to the lower second and
        // remove 2 seconds.
        // See #82: https://github.com/dadoonet/fscrawler/issues/82
        scanDate = scanDate.minus(2, ChronoUnit.SECONDS);
        FsJob fsJob = FsJob.builder()
                .setName(jobName)
                .setLastrun(scanDate)
                .setIndexed(stats.getNbDocScan())
                .setDeleted(stats.getNbDocDeleted())
                .build();
        fsJobFileHandler.write(jobName, fsJob);
    }
17144
private JobConf localizeJob(TaskInProgress tip) throws IOException {
    Path localJarFile = null;
    Task t = tip.getTask();
    JobID jobId = t.getJobID();
    Path jobFile = new Path(t.getJobFile());
    // Get sizes of JobFile and JarFile
    // sizes are -1 if they are not present.
    FileStatus status = null;
    long jobFileSize = -1;
    try {
      status = systemFS.getFileStatus(jobFile);
      jobFileSize = status.getLen();
    } catch(FileNotFoundException fe) {
      jobFileSize = -1;
    }
    Path localJobFile = lDirAlloc.getLocalPathForWrite(
                                    getLocalJobDir(jobId.toString())
                                    + Path.SEPARATOR + "job.xml",
                                    jobFileSize, fConf);
    RunningJob rjob = addTaskToJob(jobId, tip);
    synchronized (rjob.localizationLock) {
      if (rjob.localized == false) {
        // Actually start the job localization IO.
        FileSystem localFs = FileSystem.getLocal(fConf);
        // this will happen on a partial execution of localizeJob.
        // Sometimes the job.xml gets copied but copying job.jar
        // might throw out an exception
        // we should clean up and then try again
        Path jobDir = localJobFile.getParent();
        if (localFs.exists(jobDir)){
          LOG.warn("Deleting pre-existing jobDir: " + jobDir
              + " when localizeJob for tip " + tip);
          localFs.delete(jobDir, true);
          boolean b = localFs.mkdirs(jobDir);
          if (!b)
            throw new IOException("Not able to create job directory "
                + jobDir.toString());
        }
        systemFS.copyToLocalFile(jobFile, localJobFile);
        boolean cleanup = false;
        JobConf localJobConf = null;
        Path bigParamPath = null;
        if (fConf.getBoolean("mapred.input.dir.cleanup", true)) {
          bigParamPath = new Path(jobDir, "bigParam");
          int threshold = fConf.getInt("mapred.input.dir.cleanup.threshold", 1000);
          localJobConf = new JobConf(localJobFile, bigParamPath, localFs, threshold);
        } else {
          localJobConf = new JobConf(localJobFile);
        } 

        // create the 'work' directory
        // job-specific shared directory for use as scratch space
        Path workDir = lDirAlloc.getLocalPathForWrite(
            (getLocalJobDir(jobId.toString())
                + Path.SEPARATOR + "work"), fConf);
        if (!localFs.mkdirs(workDir)) {
          throw new IOException("Mkdirs failed to create "
              + workDir.toString());
        }
        System.setProperty("job.local.dir", workDir.toString());
        localJobConf.set("job.local.dir", workDir.toString());

        // copy Jar file to the local FS and unjar it.
        String jarFile = localJobConf.getJar();
        long jarFileSize = -1;
        boolean changeLocalJobConf = jarFile != null;
        if (changeLocalJobConf) {
          boolean shared =
            localJobConf.getBoolean("mapred.cache.shared.enabled", false);

          // If sharing is turned on, we already have the jarFileSize, so we
          // don't have to make another RPC call to NameNode
          Path jarFilePath = new Path(jarFile);
          if (shared) {
            try {
              jarFileSize =
                Long.parseLong(DistributedCache.
                    getSharedArchiveLength(localJobConf)[0]);
            } catch (NullPointerException npe) {
              jarFileSize = -1;
            }
          } else {
            try {
              status = systemFS.getFileStatus(jarFilePath);
              jarFileSize = status.getLen();
            } catch(FileNotFoundException fe) {
              jarFileSize = -1;
            }
          }
          // Here we check for and we check five times the size of jarFileSize
          // to accommodate for unjarring the jar file in work directory
          localJarFile = new Path(lDirAlloc.getLocalPathForWrite(
              getLocalJobDir(jobId.toString())
              + Path.SEPARATOR + "jars",
              5 * jarFileSize, fConf), "job.jar");
          if (!localFs.mkdirs(localJarFile.getParent())) {
            throw new IOException("Mkdirs failed to create jars directory ");
          }

          if (!shared) {
            // we copy the job jar to the local disk and unjar it
            // for the shared case - this is done inside TaskRunner
            systemFS.copyToLocalFile(jarFilePath, localJarFile);
            RunJar.unJar(new File(localJarFile.toString()),
                new File(localJarFile.getParent().toString()));
          }
          localJobConf.setJar(localJarFile.toString());
        }
        reconfigureLocalJobConf(localJobConf, localJobFile, tip, changeLocalJobConf);
        synchronized (rjob) {
          rjob.keepJobFiles = ((localJobConf.getKeepTaskFilesPattern() != null) ||
              localJobConf.getKeepFailedTaskFiles());
          rjob.jobConf = localJobConf;
          taskController.initializeJob(jobId);
          rjob.localized = true;
        }
      } else {
        // Even if job is localized we must update umbilical addresses
        // Only when the remote job tracker address get changed we will update the XML file. 
        // Refer to the logic of reconfigureLocalJobConf which override this function in 
        // CoronaTaskTracker.
        JobConf localJobConf = new JobConf(localJobFile);
        reconfigureLocalJobConf(localJobConf, localJobFile, tip, false);
      }
    }
    return new JobConf(rjob.jobConf);
  }
473195
public void setLocation(String location, Job job) throws IOException
    {
        conf = job.getConfiguration();
        setLocationFromUri(location);

        if (username != null && password != null)
        {
            ConfigHelper.setInputKeyspaceUserNameAndPassword(conf, username, password);
            CqlConfigHelper.setUserNameAndPassword(conf, username, password);
        }
        if (splitSize > 0)
            ConfigHelper.setInputSplitSize(conf, splitSize);
        if (partitionerClass!= null)
            ConfigHelper.setInputPartitioner(conf, partitionerClass);
        if (initHostAddress != null)
            ConfigHelper.setInputInitialAddress(conf, initHostAddress);
        if (rpcPort != null)
            ConfigHelper.setInputRpcPort(conf, rpcPort);
        if (nativePort != null)
            CqlConfigHelper.setInputNativePort(conf, nativePort);
        if (nativeCoreConnections != null)
            CqlConfigHelper.setInputCoreConnections(conf, nativeCoreConnections);
        if (nativeMaxConnections != null)
            CqlConfigHelper.setInputMaxConnections(conf, nativeMaxConnections);
        if (nativeMinSimultReqs != null)
            CqlConfigHelper.setInputMinSimultReqPerConnections(conf, nativeMinSimultReqs);
        if (nativeMaxSimultReqs != null)
            CqlConfigHelper.setInputMaxSimultReqPerConnections(conf, nativeMaxSimultReqs);
        if (nativeConnectionTimeout != null)
            CqlConfigHelper.setInputNativeConnectionTimeout(conf, nativeConnectionTimeout);
        if (nativeReadConnectionTimeout != null)
            CqlConfigHelper.setInputNativeReadConnectionTimeout(conf, nativeReadConnectionTimeout);
        if (nativeReceiveBufferSize != null)
            CqlConfigHelper.setInputNativeReceiveBufferSize(conf, nativeReceiveBufferSize);
        if (nativeSendBufferSize != null)
            CqlConfigHelper.setInputNativeSendBufferSize(conf, nativeSendBufferSize);
        if (nativeSolinger != null)
            CqlConfigHelper.setInputNativeSolinger(conf, nativeSolinger);
        if (nativeTcpNodelay != null)
            CqlConfigHelper.setInputNativeTcpNodelay(conf, nativeTcpNodelay);
        if (nativeReuseAddress != null)
            CqlConfigHelper.setInputNativeReuseAddress(conf, nativeReuseAddress);
        if (nativeKeepAlive != null)
            CqlConfigHelper.setInputNativeKeepAlive(conf, nativeKeepAlive);
        if (nativeAuthProvider != null)
            CqlConfigHelper.setInputNativeAuthProvider(conf, nativeAuthProvider);
        if (nativeSSLTruststorePath != null)
            CqlConfigHelper.setInputNativeSSLTruststorePath(conf, nativeSSLTruststorePath);
        if (nativeSSLKeystorePath != null)
            CqlConfigHelper.setInputNativeSSLKeystorePath(conf, nativeSSLKeystorePath);
        if (nativeSSLTruststorePassword != null)
            CqlConfigHelper.setInputNativeSSLTruststorePassword(conf, nativeSSLTruststorePassword);
        if (nativeSSLKeystorePassword != null)
            CqlConfigHelper.setInputNativeSSLKeystorePassword(conf, nativeSSLKeystorePassword);
        if (nativeSSLCipherSuites != null)
            CqlConfigHelper.setInputNativeSSLCipherSuites(conf, nativeSSLCipherSuites);

        ConfigHelper.setInputColumnFamily(conf, keyspace, column_family);
        setConnectionInformation();

        CqlConfigHelper.setInputCQLPageRowSize(conf, String.valueOf(pageSize));
        if (inputCql != null)
            CqlConfigHelper.setInputCql(conf, inputCql);
        if (columns != null)
            CqlConfigHelper.setInputColumns(conf, columns);
        if (whereClause != null)
            CqlConfigHelper.setInputWhereClauses(conf, whereClause);

        String whereClauseForPartitionFilter = getWhereClauseForPartitionFilter();
        String wc = whereClause != null && !whereClause.trim().isEmpty() 
                               ? whereClauseForPartitionFilter == null ? whereClause: String.format("%s AND %s", whereClause.trim(), whereClauseForPartitionFilter)
                               : whereClauseForPartitionFilter;

        if (wc != null)
        {
            logger.debug("where clause: {}", wc);
            CqlConfigHelper.setInputWhereClauses(conf, wc);
        } 
        if (System.getenv(PIG_INPUT_SPLIT_SIZE) != null)
        {
            try
            {
                ConfigHelper.setInputSplitSize(conf, Integer.parseInt(System.getenv(PIG_INPUT_SPLIT_SIZE)));
            }
            catch (NumberFormatException e)
            {
                throw new IOException("PIG_INPUT_SPLIT_SIZE is not a number", e);
            }           
        }

        if (ConfigHelper.getInputInitialAddress(conf) == null)
            throw new IOException("PIG_INPUT_INITIAL_ADDRESS or PIG_INITIAL_ADDRESS environment variable not set");
        if (ConfigHelper.getInputPartitioner(conf) == null)
            throw new IOException("PIG_INPUT_PARTITIONER or PIG_PARTITIONER environment variable not set");
        if (loadSignature == null)
            loadSignature = location;

        initSchema(loadSignature);
    }
416116
public void setJobType(int jobType) {
	String type = null;
	switch(jobType) {
	case JOBTYPE_SINGLE:
	    type = "single"; break;
	case JOBTYPE_MULTIPLE:
	    type = "multiple"; break;
	case JOBTYPE_MPI:
	    type = "mpi"; break;
	case JOBTYPE_CONDOR:
	    type = "condor"; break;
	}
	if (type != null) {
	    set("jobtype", type);
	}
    }
689378
public void store(JobInProgress job) {
    if (active && retainTime > 0) {
      JobID jobId = job.getStatus().getJobID();
      Path jobStatusFile = getInfoFilePath(jobId);
      try {
        FSDataOutputStream dataOut = fs.create(jobStatusFile);

        job.getStatus().write(dataOut);

        job.getProfile().write(dataOut);

        job.getCounters().write(dataOut);

        TaskCompletionEvent[] events = 
                job.getTaskCompletionEvents(0, Integer.MAX_VALUE);
        dataOut.writeInt(events.length);
        for (TaskCompletionEvent event : events) {
          event.write(dataOut);
        }

        dataOut.close();
      } catch (IOException ex) {
        LOG.warn("Could not store [" + jobId + "] job info : " +
                 ex.getMessage(), ex);
        try {
          fs.delete(jobStatusFile, true);
        }
        catch (IOException ex1) {
          //ignore
        }
      }
    }
  }
1143940
@Override
    public void setLocation(String location, Job job)
            throws IOException {
        // The location is assumed to be comma separated paths.
        FileInputFormat.setInputPaths(job, location);

        requiredFieldList = (RequiredFieldList) getFromUDFContext(PRUNE_PROJECTION_INFO);

        // If we encounter a PushDown Projection we strip the requestedFields to only the needed ones
        // This pruning will very effectively push the projection down into the actual parser system.
        if (requiredFieldList != null &&
            originalRequestedFields == null) { // Avoid pruning twice !!
            Set<Integer> requestedFieldIndexes = new HashSet<>();
            for (RequiredField requiredField : requiredFieldList.getFields()) {
                requestedFieldIndexes.add(requiredField.getIndex());
            }
            List<String> prunedRequestedFields = new ArrayList<>(requestedFieldIndexes.size());
            int index = 0;
            for (String field : requestedFields) {
                if (requestedFieldIndexes.contains(index)) {
                    prunedRequestedFields.add(field);
                }
                ++index;
            }
            originalRequestedFields = requestedFields;
            requestedFields = prunedRequestedFields;
        }
    }
844199
public JobStatus readJobStatus(JobID jobId) {
    JobStatus jobStatus = null;
    
    if (null == jobId) {
      LOG.warn("Could not read job status for null jobId");
      return null;
    }
    
    if (active) {
      try {
        FSDataInputStream dataIn = getJobInfoFile(jobId);
        if (dataIn != null) {
          jobStatus = readJobStatus(dataIn);
          dataIn.close();
        }
      } catch (IOException ex) {
        LOG.warn("Could not read [" + jobId + "] job status : " + ex, ex);
      }
    }
    return jobStatus;
  }
184508
@GET
  @Path(ServiceConstants.GET_STATUS)
  @JacksonFeatures(serializationEnable = {SerializationFeature.INDENT_OUTPUT})
  public Response getStatus(@QueryParam("jobId") final long jobId) {
    return RestUtils.call(new RestUtils.RestCallable<JobInfo>() {
      @Override
      public JobInfo call() throws Exception {
        return mJobMaster.getStatus(jobId);
      }
    }, ServerConfiguration.global());
  }
622955
private void writeJobState(JsonWriter jsonWriter, JobState jobState) throws IOException {
    jobState.toJson(jsonWriter, this.keepConfig);
  }
1011718
public boolean monitorAndPrintJob(JobConf conf,
                                    RunningJob job
  ) throws IOException, InterruptedException {
    String lastReport = null;
    TaskStatusFilter filter;
    filter = getTaskOutputFilter(conf);
    JobID jobId = job.getID();
    LOG.info("Running job: " + jobId);
    int eventCounter = 0;
    boolean profiling = conf.getProfileEnabled();
    Configuration.IntegerRanges mapRanges = conf.getProfileTaskRange(true);
    Configuration.IntegerRanges reduceRanges = conf.getProfileTaskRange(false);

    while (!job.isComplete()) {
      Thread.sleep(MAX_JOBPROFILE_AGE);
      String report =
        (" map " + StringUtils.formatPercent(job.mapProgress(), 0)+
            " reduce " +
            StringUtils.formatPercent(job.reduceProgress(), 0));
      if (!report.equals(lastReport)) {
        LOG.info(report);
        lastReport = report;
      }

      TaskCompletionEvent[] events =
        job.getTaskCompletionEvents(eventCounter);
      eventCounter += events.length;
      for(TaskCompletionEvent event : events){
        TaskCompletionEvent.Status status = event.getTaskStatus();
        if (profiling &&
            (status == TaskCompletionEvent.Status.SUCCEEDED ||
                status == TaskCompletionEvent.Status.FAILED) &&
                (event.isMap ? mapRanges : reduceRanges).
                isIncluded(event.idWithinJob())) {
          downloadProfile(event);
        }
        switch(filter){
        case NONE:
          break;
        case SUCCEEDED:
          if (event.getTaskStatus() ==
            TaskCompletionEvent.Status.SUCCEEDED){
            LOG.info(event.toString());
            displayTaskLogs(event.getTaskAttemptId(), event.getTaskTrackerHttp());
          }
          break;
        case FAILED:
          if (event.getTaskStatus() ==
            TaskCompletionEvent.Status.FAILED){
            LOG.info(event.toString());
            // Displaying the task diagnostic information
            TaskAttemptID taskId = event.getTaskAttemptId();
            String[] taskDiagnostics =
              jobSubmitClient.getTaskDiagnostics(taskId);
            if (taskDiagnostics != null) {
              for(String diagnostics : taskDiagnostics){
                System.err.println(diagnostics);
              }
            }
            // Displaying the task logs
            displayTaskLogs(event.getTaskAttemptId(), event.getTaskTrackerHttp());
          }
          break;
        case KILLED:
          if (event.getTaskStatus() == TaskCompletionEvent.Status.KILLED){
            LOG.info(event.toString());
          }
          break;
        case ALL:
          LOG.info(event.toString());
          displayTaskLogs(event.getTaskAttemptId(), event.getTaskTrackerHttp());
          break;
        }
      }
    }
    LOG.info("Job complete: " + jobId);
    Counters counters = job.getCounters();
    if (counters != null) {
      counters.log(LOG);
    }
    return job.isSuccessful();
  }
102291
private void writeObject(ObjectOutputStream out) throws IOException {
		super.write(out);
		out.writeUTF(mapredOutputFormat.getClass().getName());
		jobConf.write(out);
	}
712300
protected List<FileStatus> listStatus(JobContext job) throws IOException
    {
        List<FileStatus> result = new ArrayList<FileStatus>();
        Path[] dirs = getInputPaths(job);
        if (dirs.length == 0) {
            throw new IOException("No input paths specified in job");
        }

        // Get tokens for all the required FileSystems..
        TokenCache.obtainTokensForNamenodes(job.getCredentials(), dirs, job.getConfiguration());

        List<IOException> errors = new ArrayList<IOException>();
        for (Path p : dirs) {
            FileSystem fs = p.getFileSystem(job.getConfiguration());
            final SmilePathFilter filter = new SmilePathFilter();
            FileStatus[] matches = fs.globStatus(p, filter);
            if (matches == null) {
                errors.add(new IOException("Input path does not exist: " + p));
            }
            else if (matches.length == 0) {
                errors.add(new IOException("Input Pattern " + p + " matches 0 files"));
            }
            else {
                for (FileStatus globStat : matches) {
                    if (globStat.isDir()) {
                        Collections.addAll(result, fs.listStatus(globStat.getPath(), filter));
                    }
                    else {
                        result.add(globStat);
                    }
                }
            }
        }

        if (!errors.isEmpty()) {
            throw new InvalidInputException(errors);
        }

        return result;
    }
535257
public static boolean runJob(JobConf job) throws IOException {
    JobClient jc = new JobClient(job);
    boolean sucess = true;
    RunningJob running = null;
    try {
      running = jc.submitJob(job);
      JobID jobId = running.getID();
      System.out.println("Job " + jobId + " is submitted");
      while (!running.isComplete()) {
        System.out.println("Job " + jobId + " is still running.");
        try {
          Thread.sleep(60000);
        } catch (InterruptedException e) {
        }
        running = jc.getJob(jobId);
      }
      sucess = running.isSuccessful();
    } finally {
      if (!sucess && (running != null)) {
        running.killJob();
      }
      jc.close();
    }
    return sucess;
  }