code: @Override
    public JobDesc create(QualifiedJobId qualifiedJobId, long submitTimeMillis,
        Configuration jobConf) {
      String appId = getAppId(jobConf);
      String version = jobConf.get(Constants.PIG_VERSION_CONF_KEY,
          Constants.UNKNOWN);
      long pigSubmitTimeMillis = jobConf.getLong(Constants.PIG_RUN_CONF_KEY, 0);
  
      // This means that Constants.PIG_RUN_CONF_KEY was not present (for jobs
      // launched with an older pig version).
      if (pigSubmitTimeMillis == 0) {
        String pigLogfile = jobConf.get(Constants.PIG_LOG_FILE_CONF_KEY);
        if (pigLogfile == null) {
          // Should be rare, but we're seeing this happen occasionally
          // Give up on grouping the jobs within the run together, and treat these as individual runs.
          pigSubmitTimeMillis = submitTimeMillis;
        } else {
          pigSubmitTimeMillis = getScriptStartTimeFromLogfileName(pigLogfile);
        }
      }
  
      return create(qualifiedJobId, jobConf, appId, version, Framework.PIG,
          pigSubmitTimeMillis);
    }
docs: /*
  (non-Javadoc)
  
  @see
  com.twitter.hraven.JobKeyFactoryBase#create(com.twitter.corestorage
  .rhaven.QualifiedJobId, long, org.apache.hadoop.conf.Configuration)