private void submitJob(PigJob job) {
    String date = new SimpleDateFormat("dd-MM-yyyy-HH-mm-ss").format(new Date());
    String jobsBaseDir = context.getProperties().get("jobs.dir");
    String storeBaseDir = context.getProperties().get("store.dir");
    if (storeBaseDir == null || storeBaseDir.compareTo("null") == 0 || storeBaseDir.compareTo("") == 0)
      storeBaseDir = context.getProperties().get("jobs.dir");
    String jobNameCleaned = job.getTitle().toLowerCase().replaceAll("[^a-zA-Z0-9 ]+", "").replace(" ", "_");
    String storedir = String.format(storeBaseDir + "/%s_%s", jobNameCleaned, date);
    String statusdir = String.format(jobsBaseDir + "/%s_%s", jobNameCleaned, date);

    String newPigScriptPath = storedir + "/script.pig";
    String newSourceFilePath = storedir + "/source.pig";
    String newPythonScriptPath = storedir + "/udf.py";
    String templetonParamsFilePath = storedir + "/params";

    try {
      // additional file can be passed to copy into work directory
      if (job.getSourceFileContent() != null && !job.getSourceFileContent().isEmpty()) {
        String sourceFileContent = job.getSourceFileContent();
        job.setSourceFileContent(null); // we should not store content in DB
        save(job);

        FSDataOutputStream stream = UserLocalObjects.getHdfsApi(context).create(newSourceFilePath, true);
        stream.writeBytes(sourceFileContent);
        stream.close();
      } else {
        if (job.getSourceFile() != null && !job.getSourceFile().isEmpty()) {
          // otherwise, just copy original file
          UserLocalObjects.getHdfsApi(context).copy(job.getSourceFile(), newSourceFilePath);
        }
      }
    } catch (IOException e) {
      throw new ServiceFormattedException("Can't create/copy source file: " + e.toString(), e);
    } catch (InterruptedException e) {
      throw new ServiceFormattedException("Can't create/copy source file: " + e.toString(), e);
    } catch (HdfsApiException e) {
      throw new ServiceFormattedException("Can't copy source file from " + job.getSourceFile() +
          " to " + newPigScriptPath, e);
    }

    try {
      // content can be passed from front-end with substituted arguments
      if (job.getForcedContent() != null && !job.getForcedContent().isEmpty()) {
        String forcedContent = job.getForcedContent();
        String defaultUrl = context.getProperties().get("webhdfs.url");
        URI uri = new URI(defaultUrl);

        // variable for sourceFile can be passed from front-end
        // cannot use webhdfs url as webhcat does not support http authentication
        // using url hdfs://host/sourcefilepath
        if(uri.getScheme().equals("webhdfs")){
          defaultUrl = "hdfs://" + uri.getHost();
        }

        forcedContent = forcedContent.replace("${sourceFile}",
            defaultUrl + newSourceFilePath);
        job.setForcedContent(null); // we should not store content in DB
        save(job);

        FSDataOutputStream stream = UserLocalObjects.getHdfsApi(context).create(newPigScriptPath, true);
        stream.writeBytes(forcedContent);
        stream.close();
      } else {
        // otherwise, just copy original file
        UserLocalObjects.getHdfsApi(context).copy(job.getPigScript(), newPigScriptPath);
      }
    } catch (HdfsApiException e) {
      throw new ServiceFormattedException("Can't copy pig script file from " + job.getPigScript() +
          " to " + newPigScriptPath, e);
    } catch (IOException e) {
      throw new ServiceFormattedException("Can't create/copy pig script file: " + e.getMessage(), e);
    } catch (InterruptedException e) {
      throw new ServiceFormattedException("Can't create/copy pig script file: " + e.getMessage(), e);
    } catch (URISyntaxException e) {
      throw new ServiceFormattedException("Can't create/copy pig script file: " + e.getMessage(), e);
    }

    if (job.getPythonScript() != null && !job.getPythonScript().isEmpty()) {
      try {
        UserLocalObjects.getHdfsApi(context).copy(job.getPythonScript(), newPythonScriptPath);
      } catch (HdfsApiException e) {
        throw new ServiceFormattedException("Can't copy python udf script file from " + job.getPythonScript() +
            " to " + newPythonScriptPath);
      } catch (IOException e) {
        throw new ServiceFormattedException("Can't create/copy python udf file: " + e.toString(), e);
      } catch (InterruptedException e) {
        throw new ServiceFormattedException("Can't create/copy python udf file: " + e.toString(), e);
      }
    }

    try {
      FSDataOutputStream stream = UserLocalObjects.getHdfsApi(context).create(templetonParamsFilePath, true);
      if (job.getTempletonArguments() != null) {
        stream.writeBytes(job.getTempletonArguments());
      }
      stream.close();
    } catch (IOException e) {
      throw new ServiceFormattedException("Can't create params file: " + e.toString(), e);
    } catch (InterruptedException e) {
      throw new ServiceFormattedException("Can't create params file: " + e.toString(), e);
    }
    job.setPigScript(newPigScriptPath);

    job.setStatusDir(statusdir);
    job.setDateStarted(System.currentTimeMillis() / 1000L);

    TempletonApi.JobData data;
    try {
      data = UserLocalObjects.getTempletonApi(context).runPigQuery(new File(job.getPigScript()), statusdir, job.getTempletonArguments());
      if (data.id != null) {
        job.setJobId(data.id);
        JobPolling.pollJob(this, job);
      } else {
        throw new AmbariApiException("Cannot get id for the Job.");
      }
    } catch (IOException templetonBadResponse) {
      String msg = String.format("Templeton bad response: %s", templetonBadResponse.toString());
      LOG.debug(msg);
      throw new ServiceFormattedException(msg, templetonBadResponse);
    }
  }
Running job
@param job job bean
contrib/views/pig/src/main/java/org/apache/ambari/view/pig/resources/jobs/JobResourceManager.java