<commit_before>package io.hops.hopsworks.common.jobs.spark;

import com.google.common.base.Strings;
import io.hops.hopsworks.common.dao.jobs.description.Jobs;
import io.hops.hopsworks.common.hdfs.DistributedFileSystemOps;
import io.hops.hopsworks.common.jobs.AsynchronousJobExecutor;
import io.hops.hopsworks.common.jobs.jobhistory.JobType;
import io.hops.hopsworks.common.jobs.yarn.LocalResourceDTO;
import io.hops.hopsworks.common.jobs.yarn.ServiceProperties;
import io.hops.hopsworks.common.jobs.yarn.YarnRunner;
import io.hops.hopsworks.common.util.Settings;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.client.api.YarnClient;

/**
 * Builder class for a Spark YarnRunner. Implements the common logic needed
 * for any Spark job to be started and builds a YarnRunner instance.
 * <p>
 */
public class SparkYarnRunnerBuilder {

  private static final Logger LOG = Logger.getLogger(
      SparkYarnRunnerBuilder.class.getName());

  //Necessary parameters
  private final Jobs job;

  //Optional parameters
  private final List<String> jobArgs = new ArrayList<>();
  private String jobName = "Untitled Spark Job";
  private final List<LocalResourceDTO> extraFiles = new ArrayList<>();
  private int numberOfExecutors = 1;
  private int numberOfExecutorsMin = Settings.SPARK_MIN_EXECS;
  private int numberOfExecutorsMax = Settings.SPARK_MAX_EXECS;
  private int numberOfExecutorsInit = Settings.SPARK_INIT_EXECS;
  private int executorCores = 1;
  private String properties;
  private boolean dynamicExecutors;
  private String executorMemory = "512m";
  private int driverMemory = 1024; // in MB
  private int driverCores = 1;
  private String driverQueue;
  private int numOfGPUs = 0;
  private int numOfPs = 0;
  private final Map<String, String> envVars = new HashMap<>();
  private final Map<String, String> sysProps = new HashMap<>();
  private String classPath;
  private ServiceProperties serviceProps;
  private String sessionId;
  final private Set<String> blacklistedProps = new HashSet<>();

  public SparkYarnRunnerBuilder(Jobs job) {
    this.job = job;
    SparkJobConfiguration jobConfig = (SparkJobConfiguration) job.
        getJobConfig();

    if (jobConfig.getAppPath() == null || jobConfig.getAppPath().isEmpty()) {
      throw new IllegalArgumentException(
          "Path to application executable cannot be empty!");
    }
    if (jobConfig.getMainClass() == null || jobConfig.getMainClass().isEmpty()) {
      throw new IllegalArgumentException(
          "Name of the main class cannot be empty!");
    }

  }

  /**
   * Get a YarnRunner instance that will launch a Spark job.
   *
   * @param project name of the project
   * @param sparkUser
   * @param jobUser
   * @param sparkDir
   * @param services
   * @param settings
   * @return The YarnRunner instance to launch the Spark job on Yarn.
   * @throws IOException If creation failed.
   */
  public YarnRunner getYarnRunner(String project, String sparkUser,
      String jobUser, final String sparkDir, AsynchronousJobExecutor services,
      final DistributedFileSystemOps dfsClient, final YarnClient yarnClient,
      Settings settings)
      throws IOException {

    //Read blacklisted properties from local spark dir
    File blacklist = new File(sparkDir + "/" + Settings.SPARK_BLACKLISTED_PROPS);
    try (InputStream is = new FileInputStream(blacklist)) {
      byte[] data = new byte[(int) blacklist.length()];
      is.read(data);
      String content = new String(data, "UTF-8");
      blacklistedProps.addAll(Arrays.asList(content.split("\n")));
    }

    JobType jobType = ((SparkJobConfiguration) job.getJobConfig()).
        getType();
    String appPath = ((SparkJobConfiguration) job.getJobConfig()).
        getAppPath();

    String hdfsSparkJarPath = Settings.getHdfsSparkJarPath(sparkUser);
    String log4jPath = Settings.getSparkLog4JPath(sparkUser);
    StringBuilder pythonPath = null;
    StringBuilder pythonPathExecs = null;
    //Create a builder
    YarnRunner.Builder builder = new YarnRunner.Builder(Settings.SPARK_AM_MAIN);
    builder.setJobType(jobType);
    builder.setYarnClient(yarnClient);
    builder.setDfsClient(dfsClient);
    builder.setJobUser(jobUser);
    try {
      String password = services.getBaseHadoopClientsService()
          .getProjectSpecificUserCertPassword(jobUser);
      builder.setKeyStorePassword(password);
      builder.setTrustStorePassword(password);
    } catch (Exception ex) {
      throw new IOException(ex);
    }

    String stagingPath = "/Projects/ " + project + "/"
        + Settings.PROJECT_STAGING_DIR + "/.sparkjobstaging";
    builder.localResourcesBasePath(stagingPath);

    builder.addLocalResource(new LocalResourceDTO(
        Settings.SPARK_LOCALIZED_LIB_DIR, hdfsSparkJarPath,
        LocalResourceVisibility.PRIVATE.toString(),
        LocalResourceType.ARCHIVE.toString(), null), false);
    //Add log4j
    builder.addLocalResource(new LocalResourceDTO(
        Settings.SPARK_LOG4J_PROPERTIES, log4jPath,
        LocalResourceVisibility.APPLICATION.toString(),
        LocalResourceType.FILE.toString(), null), false);
    //Add metrics 
    builder.addLocalResource(new LocalResourceDTO(
        Settings.SPARK_METRICS_PROPERTIES, settings.getSparkConfDir() + "/metrics.properties",
        LocalResourceVisibility.PRIVATE.toString(),
        LocalResourceType.FILE.toString(), null), false);

    //Add app file
    String appExecName = null;
    if (jobType == JobType.SPARK) {
      appExecName = Settings.SPARK_LOCRSC_APP_JAR;<commit_msg>Remove stray white-space character from spark staging directory<commit_after>package io.hops.hopsworks.common.jobs.spark;

import com.google.common.base.Strings;
import io.hops.hopsworks.common.dao.jobs.description.Jobs;
import io.hops.hopsworks.common.hdfs.DistributedFileSystemOps;
import io.hops.hopsworks.common.jobs.AsynchronousJobExecutor;
import io.hops.hopsworks.common.jobs.jobhistory.JobType;
import io.hops.hopsworks.common.jobs.yarn.LocalResourceDTO;
import io.hops.hopsworks.common.jobs.yarn.ServiceProperties;
import io.hops.hopsworks.common.jobs.yarn.YarnRunner;
import io.hops.hopsworks.common.util.Settings;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.client.api.YarnClient;

/**
 * Builder class for a Spark YarnRunner. Implements the common logic needed
 * for any Spark job to be started and builds a YarnRunner instance.
 * <p>
 */
public class SparkYarnRunnerBuilder {

  private static final Logger LOG = Logger.getLogger(
      SparkYarnRunnerBuilder.class.getName());

  //Necessary parameters
  private final Jobs job;

  //Optional parameters
  private final List<String> jobArgs = new ArrayList<>();
  private String jobName = "Untitled Spark Job";
  private final List<LocalResourceDTO> extraFiles = new ArrayList<>();
  private int numberOfExecutors = 1;
  private int numberOfExecutorsMin = Settings.SPARK_MIN_EXECS;
  private int numberOfExecutorsMax = Settings.SPARK_MAX_EXECS;
  private int numberOfExecutorsInit = Settings.SPARK_INIT_EXECS;
  private int executorCores = 1;
  private String properties;
  private boolean dynamicExecutors;
  private String executorMemory = "512m";
  private int driverMemory = 1024; // in MB
  private int driverCores = 1;
  private String driverQueue;
  private int numOfGPUs = 0;
  private int numOfPs = 0;
  private final Map<String, String> envVars = new HashMap<>();
  private final Map<String, String> sysProps = new HashMap<>();
  private String classPath;
  private ServiceProperties serviceProps;
  private String sessionId;
  final private Set<String> blacklistedProps = new HashSet<>();

  public SparkYarnRunnerBuilder(Jobs job) {
    this.job = job;
    SparkJobConfiguration jobConfig = (SparkJobConfiguration) job.
        getJobConfig();

    if (jobConfig.getAppPath() == null || jobConfig.getAppPath().isEmpty()) {
      throw new IllegalArgumentException(
          "Path to application executable cannot be empty!");
    }
    if (jobConfig.getMainClass() == null || jobConfig.getMainClass().isEmpty()) {
      throw new IllegalArgumentException(
          "Name of the main class cannot be empty!");
    }

  }

  /**
   * Get a YarnRunner instance that will launch a Spark job.
   *
   * @param project name of the project
   * @param sparkUser
   * @param jobUser
   * @param sparkDir
   * @param services
   * @param settings
   * @return The YarnRunner instance to launch the Spark job on Yarn.
   * @throws IOException If creation failed.
   */
  public YarnRunner getYarnRunner(String project, String sparkUser,
      String jobUser, final String sparkDir, AsynchronousJobExecutor services,
      final DistributedFileSystemOps dfsClient, final YarnClient yarnClient,
      Settings settings)
      throws IOException {

    //Read blacklisted properties from local spark dir
    File blacklist = new File(sparkDir + "/" + Settings.SPARK_BLACKLISTED_PROPS);
    try (InputStream is = new FileInputStream(blacklist)) {
      byte[] data = new byte[(int) blacklist.length()];
      is.read(data);
      String content = new String(data, "UTF-8");
      blacklistedProps.addAll(Arrays.asList(content.split("\n")));
    }

    JobType jobType = ((SparkJobConfiguration) job.getJobConfig()).
        getType();
    String appPath = ((SparkJobConfiguration) job.getJobConfig()).
        getAppPath();

    String hdfsSparkJarPath = Settings.getHdfsSparkJarPath(sparkUser);
    String log4jPath = Settings.getSparkLog4JPath(sparkUser);
    StringBuilder pythonPath = null;
    StringBuilder pythonPathExecs = null;
    //Create a builder
    YarnRunner.Builder builder = new YarnRunner.Builder(Settings.SPARK_AM_MAIN);
    builder.setJobType(jobType);
    builder.setYarnClient(yarnClient);
    builder.setDfsClient(dfsClient);
    builder.setJobUser(jobUser);
    try {
      String password = services.getBaseHadoopClientsService()
          .getProjectSpecificUserCertPassword(jobUser);
      builder.setKeyStorePassword(password);
      builder.setTrustStorePassword(password);
    } catch (Exception ex) {
      throw new IOException(ex);
    }

    String stagingPath = "/Projects/" + project + "/"
        + Settings.PROJECT_STAGING_DIR + "/.sparkjobstaging";
    builder.localResourcesBasePath(stagingPath);

    builder.addLocalResource(new LocalResourceDTO(
        Settings.SPARK_LOCALIZED_LIB_DIR, hdfsSparkJarPath,
        LocalResourceVisibility.PRIVATE.toString(),
        LocalResourceType.ARCHIVE.toString(), null), false);
    //Add log4j
    builder.addLocalResource(new LocalResourceDTO(
        Settings.SPARK_LOG4J_PROPERTIES, log4jPath,
        LocalResourceVisibility.APPLICATION.toString(),
        LocalResourceType.FILE.toString(), null), false);
    //Add metrics 
    builder.addLocalResource(new LocalResourceDTO(
        Settings.SPARK_METRICS_PROPERTIES, settings.getSparkConfDir() + "/metrics.properties",
        LocalResourceVisibility.PRIVATE.toString(),
        LocalResourceType.FILE.toString(), null), false);

    //Add app file
    String appExecName = null;
    if (jobType == JobType.SPARK) {
      appExecName = Settings.SPARK_LOCRSC_APP_JAR;
{'fraction_non_alphanumeric': 0.0969611658524468, 'fraction_numerical': 0.002980872733294692, 'mean_word_length': 4.091905564924114, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6966043', 'n_tokens_mistral': 3591, 'n_tokens_neox': 3481, 'n_words': 850}