<commit_before>package se.kth.bbc.jobs.spark;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import se.kth.bbc.jobs.yarn.YarnRunner;
import se.kth.hopsworks.controller.LocalResourceDTO;
import se.kth.hopsworks.util.Settings;

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

  //Necessary parameters
  private final String appJarPath, mainClass;

  //Optional parameters
  private final List<String> jobArgs = new ArrayList<>();
  private String jobName = "Untitled Spark Job";
  private List<LocalResourceDTO> extraFiles = new ArrayList<>();

  private int numberOfExecutors = 1;
  private int executorCores = 1;
  private String executorMemory = "512m";
  private int driverMemory = 1024; // in MB
  private int driverCores = 1;
  private String driverQueue;
  private final Map<String, String> envVars = new HashMap<>();
  private final Map<String, String> sysProps = new HashMap<>();
  private String classPath;
  private String sparkHistoryServerIp;

  public SparkYarnRunnerBuilder(String appJarPath, String mainClass) {
    if (appJarPath == null || appJarPath.isEmpty()) {
      throw new IllegalArgumentException(
              "Path to application jar cannot be empty!");
    }
    if (mainClass == null || mainClass.isEmpty()) {
      throw new IllegalArgumentException(
              "Name of the main class cannot be empty!");
    }
    this.appJarPath = appJarPath;
    this.mainClass = mainClass;
  }

  /**
   * Get a YarnRunner instance that will launch a Spark job.
   * <p/>
   * @param project name of the project
   * @param sparkUser
   * @param hadoopDir
   * @param sparkDir
   * @param nameNodeIpPort
   * @return The YarnRunner instance to launch the Spark job on Yarn.
   * @throws IOException If creation failed.
   */
  public YarnRunner getYarnRunner(String project, String sparkUser,
          final String hadoopDir, final String sparkDir, final String nameNodeIpPort)
          throws IOException {

    String sparkClasspath = Settings.getSparkDefaultClasspath(sparkDir);
    String hdfsSparkJarPath = Settings.getHdfsSparkJarPath(sparkUser);

    //TODO: include driver memory as am memory
    //Create a builder
    YarnRunner.Builder builder = new YarnRunner.Builder(Settings.SPARK_AM_MAIN);

    //Set Spark staging directory
//    String stagingPath = File.separator + "user" + File.separator + Utils.
//            getYarnUser() + File.separator + Settings.SPARK_STAGING_DIR
//            + File.separator + YarnRunner.APPID_PLACEHOLDER;
    String stagingPath = File.separator + "Projects" + File.separator + project
            + File.separator
            + Settings.PROJECT_STAGING_DIR + File.separator
            + YarnRunner.APPID_PLACEHOLDER;
    builder.localResourcesBasePath(stagingPath);

    //Add Spark jar
    builder.addLocalResource(new LocalResourceDTO(
            Settings.SPARK_LOCRSC_SPARK_JAR, hdfsSparkJarPath,
            LocalResourceVisibility.PUBLIC.toString(), 
            LocalResourceType.FILE.toString(), null), false);
    //Add app jar
    builder.addLocalResource(new LocalResourceDTO(
            Settings.SPARK_LOCRSC_APP_JAR, appJarPath, 
            LocalResourceVisibility.PUBLIC.toString(), 
            LocalResourceType.FILE.toString(), null), 
            !appJarPath.startsWith("hdfs:"));

    //Add extra files to local resources, use filename as key
    for (LocalResourceDTO dto : extraFiles) {
            builder.addLocalResource(dto, false);
    }

    //Set Spark specific environment variables
    builder.addToAppMasterEnvironment("SPARK_YARN_MODE", "true");
    builder.addToAppMasterEnvironment("SPARK_YARN_STAGING_DIR", stagingPath);
    builder.addToAppMasterEnvironment("SPARK_USER", sparkUser);
//    builder.addToAppMasterEnvironment("SPARK_USER", );
    // TODO - Change spark user here
//    builder.addToAppMasterEnvironment("SPARK_USER", Utils.getYarnUser());
    if (classPath == null || classPath.isEmpty()) {
      builder.addToAppMasterEnvironment("CLASSPATH", sparkClasspath);
    } else {
      builder.addToAppMasterEnvironment("CLASSPATH", classPath + ":"
              + sparkClasspath);
    }
    for (String key : envVars.keySet()) {
      builder.addToAppMasterEnvironment(key, envVars.get(key));
    }

    for (String s : sysProps.keySet()) {
      String option = escapeForShell("-D" + s + "=" + sysProps.get(s));
      builder.addJavaOption(option);
    }

    //Add local resources to spark environment too
    builder.addCommand(new SparkSetEnvironmentCommand());

    //Set up command
    StringBuilder amargs = new StringBuilder("--class ");
    amargs.append(mainClass);

    // spark 1.5.x replaced --num-executors with --properties-file
    // https://fossies.org/diffs/spark/1.4.1_vs_1.5.0/
    // amargs.append(" --num-executors ").append(numberOfExecutors);
    amargs.append(" --executor-cores ").append(executorCores);
    amargs.append(" --executor-memory ").append(executorMemory);
    
    for (String s : jobArgs) {
      amargs.append(" --arg ").append(s);
    }
    builder.amArgs(amargs.toString());

    //Set up Yarn properties
    builder.amMemory(driverMemory);
    builder.amVCores(driverCores);<commit_msg>Removed local spark classpath.<commit_after>package se.kth.bbc.jobs.spark;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import se.kth.bbc.jobs.yarn.YarnRunner;
import se.kth.hopsworks.controller.LocalResourceDTO;
import se.kth.hopsworks.util.Settings;

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

  //Necessary parameters
  private final String appJarPath, mainClass;

  //Optional parameters
  private final List<String> jobArgs = new ArrayList<>();
  private String jobName = "Untitled Spark Job";
  private List<LocalResourceDTO> extraFiles = new ArrayList<>();

  private int numberOfExecutors = 1;
  private int executorCores = 1;
  private String executorMemory = "512m";
  private int driverMemory = 1024; // in MB
  private int driverCores = 1;
  private String driverQueue;
  private final Map<String, String> envVars = new HashMap<>();
  private final Map<String, String> sysProps = new HashMap<>();
  private String classPath;
  private String sparkHistoryServerIp;

  public SparkYarnRunnerBuilder(String appJarPath, String mainClass) {
    if (appJarPath == null || appJarPath.isEmpty()) {
      throw new IllegalArgumentException(
              "Path to application jar cannot be empty!");
    }
    if (mainClass == null || mainClass.isEmpty()) {
      throw new IllegalArgumentException(
              "Name of the main class cannot be empty!");
    }
    this.appJarPath = appJarPath;
    this.mainClass = mainClass;
  }

  /**
   * Get a YarnRunner instance that will launch a Spark job.
   * <p/>
   * @param project name of the project
   * @param sparkUser
   * @param hadoopDir
   * @param sparkDir
   * @param nameNodeIpPort
   * @return The YarnRunner instance to launch the Spark job on Yarn.
   * @throws IOException If creation failed.
   */
  public YarnRunner getYarnRunner(String project, String sparkUser,
          final String hadoopDir, final String sparkDir, final String nameNodeIpPort)
          throws IOException {

    String sparkClasspath = Settings.getSparkDefaultClasspath(sparkDir);
    String hdfsSparkJarPath = Settings.getHdfsSparkJarPath(sparkUser);

    //TODO: include driver memory as am memory
    //Create a builder
    YarnRunner.Builder builder = new YarnRunner.Builder(Settings.SPARK_AM_MAIN);

    //Set Spark staging directory
//    String stagingPath = File.separator + "user" + File.separator + Utils.
//            getYarnUser() + File.separator + Settings.SPARK_STAGING_DIR
//            + File.separator + YarnRunner.APPID_PLACEHOLDER;
    String stagingPath = File.separator + "Projects" + File.separator + project
            + File.separator
            + Settings.PROJECT_STAGING_DIR + File.separator
            + YarnRunner.APPID_PLACEHOLDER;
    builder.localResourcesBasePath(stagingPath);

    //Add Spark jar
    builder.addLocalResource(new LocalResourceDTO(
            Settings.SPARK_LOCRSC_SPARK_JAR, hdfsSparkJarPath,
            LocalResourceVisibility.PUBLIC.toString(), 
            LocalResourceType.FILE.toString(), null), false);
    //Add app jar
    builder.addLocalResource(new LocalResourceDTO(
            Settings.SPARK_LOCRSC_APP_JAR, appJarPath, 
            LocalResourceVisibility.PUBLIC.toString(), 
            LocalResourceType.FILE.toString(), null), 
            !appJarPath.startsWith("hdfs:"));

    //Add extra files to local resources, use filename as key
    for (LocalResourceDTO dto : extraFiles) {
            builder.addLocalResource(dto, false);
    }

    //Set Spark specific environment variables
    builder.addToAppMasterEnvironment("SPARK_YARN_MODE", "true");
    builder.addToAppMasterEnvironment("SPARK_YARN_STAGING_DIR", stagingPath);
    builder.addToAppMasterEnvironment("SPARK_USER", sparkUser);
//    builder.addToAppMasterEnvironment("SPARK_USER", );
    // TODO - Change spark user here
//    builder.addToAppMasterEnvironment("SPARK_USER", Utils.getYarnUser());
//    if (classPath == null || classPath.isEmpty()) {
//      builder.addToAppMasterEnvironment("CLASSPATH", sparkClasspath);
//    } else {
//      builder.addToAppMasterEnvironment("CLASSPATH", classPath + ":"
//              + sparkClasspath);
//    }
    for (String key : envVars.keySet()) {
      builder.addToAppMasterEnvironment(key, envVars.get(key));
    }

    for (String s : sysProps.keySet()) {
      String option = escapeForShell("-D" + s + "=" + sysProps.get(s));
      builder.addJavaOption(option);
    }

    //Add local resources to spark environment too
    builder.addCommand(new SparkSetEnvironmentCommand());

    //Set up command
    StringBuilder amargs = new StringBuilder("--class ");
    amargs.append(mainClass);

    // spark 1.5.x replaced --num-executors with --properties-file
    // https://fossies.org/diffs/spark/1.4.1_vs_1.5.0/
    // amargs.append(" --num-executors ").append(numberOfExecutors);
    amargs.append(" --executor-cores ").append(executorCores);
    amargs.append(" --executor-memory ").append(executorMemory);
    
    for (String s : jobArgs) {
      amargs.append(" --arg ").append(s);
    }
    builder.amArgs(amargs.toString());

    //Set up Yarn properties
    builder.amMemory(driverMemory);
    builder.amVCores(driverCores);
{'fraction_non_alphanumeric': 0.10707576706324358, 'fraction_numerical': 0.003220323821450935, 'mean_word_length': 3.3860337387210673, 'pattern_counts': {'":': 2, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 2, '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': '2891595', 'n_tokens_mistral': 3356, 'n_tokens_neox': 3259, 'n_words': 802}