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

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import se.kth.bbc.jobs.jobhistory.JobType;
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;
  private String sessionId;//used by Kafka
  private String kafkaAddress;
  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);
    String hdfsSparkJarPath = "hdfs://10.0.2.15:8020/user/glassfish/spark.jar";
    
    //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 - old
    //builder.addLocalResource(Settings.SPARK_LOCRSC_SPARK_JAR, "hdfs://10.0.2.15:8020/user/glassfish/spark.jar",false);

    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) {
        if(dto.getName().equals(Settings.KAFKA_K_CERTIFICATE) ||
              dto.getName().equals(Settings.KAFKA_T_CERTIFICATE)){
            //TODO: Change to true, so that certs are removed
            //Currently a FileNotFound is thrown when trying to delete the file
            builder.addLocalResource(dto, false);
        } else{
            builder.addLocalResource(dto, !appJarPath.startsWith("hdfs:"));
        }
    }
  

    //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());

      //Removed local Spark classpath

//    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));
    }

    addSystemProperty(Settings.KAFKA_SESSIONID_ENV_VAR, sessionId);
    addSystemProperty(Settings.KAFKA_BROKERADDR_ENV_VAR, kafkaAddress);
    addSystemProperty(Settings.SPARK_HISTORY_SERVER_ENV, sparkHistoryServerIp);
    addSystemProperty(Settings.SPARK_NUMBER_EXECUTORS, Integer.toString(
            numberOfExecutors));

    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());

    InputStream is = null;

    try {
        is = new FileInputStream("/srv/spark/conf/spark-defaults.conf");
    } catch (FileNotFoundException e) {}
    
    //Set up command
    StringBuilder amargs = new StringBuilder("--class ");
    amargs.append(mainClass);
    
    Properties sparkProperties = new Properties();
    sparkProperties.load(is);
    
    sparkProperties.setProperty("spark.driver.memory", Integer.toString(driverMemory)+"m");
    sparkProperties.setProperty("spark.driver.cores", Integer.toString(driverCores));
    sparkProperties.setProperty("spark.executor.memory", executorMemory);
    sparkProperties.setProperty("spark.executor.cores", Integer.toString(executorCores));
    sparkProperties.setProperty("spark.executor.instances", Integer.toString(numberOfExecutors));
    
    Date date= new Date();
    Timestamp tm = new Timestamp(date.getTime());
    long timeString = tm.getTime();
    String sparkConfTempFile = "/tmp/spark-conf-dir/spark-conf_" + timeString + ".conf"; 
    
    File file = new File(sparkConfTempFile);
    FileOutputStream fop = new FileOutputStream(file);
    
    sparkProperties.store(fop, "Store Spark Properties file.");
    
    is.close();
    fop.close();
    
    builder.addFilesToRemove(sparkConfTempFile);
    
    amargs.append(" --properties-file ");
    amargs.append(sparkConfTempFile);
    
    // 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);
    builder.amQueue(driverQueue);

    //Set app name
    builder.appName(jobName);<commit_msg>skip spark config file creation<commit_after>package se.kth.bbc.jobs.spark;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import se.kth.bbc.jobs.jobhistory.JobType;
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;
  private String sessionId;//used by Kafka
  private String kafkaAddress;
  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);
    String hdfsSparkJarPath = "hdfs://10.0.2.15:8020/user/glassfish/spark.jar";
    
    //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 - old
    //builder.addLocalResource(Settings.SPARK_LOCRSC_SPARK_JAR, "hdfs://10.0.2.15:8020/user/glassfish/spark.jar",false);

    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) {
        if(dto.getName().equals(Settings.KAFKA_K_CERTIFICATE) ||
              dto.getName().equals(Settings.KAFKA_T_CERTIFICATE)){
            //TODO: Change to true, so that certs are removed
            //Currently a FileNotFound is thrown when trying to delete the file
            builder.addLocalResource(dto, false);
        } else{
            builder.addLocalResource(dto, !appJarPath.startsWith("hdfs:"));
        }
    }
  

    //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());

      //Removed local Spark classpath

//    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));
    }

    addSystemProperty(Settings.KAFKA_SESSIONID_ENV_VAR, sessionId);
    addSystemProperty(Settings.KAFKA_BROKERADDR_ENV_VAR, kafkaAddress);
    addSystemProperty(Settings.SPARK_HISTORY_SERVER_ENV, sparkHistoryServerIp);
    addSystemProperty(Settings.SPARK_NUMBER_EXECUTORS, Integer.toString(
            numberOfExecutors));
    
    List<String> jobSpecificProperties = new ArrayList<>();
    jobSpecificProperties.add(Settings.KAFKA_SESSIONID_ENV_VAR);
    jobSpecificProperties.add(Settings.KAFKA_BROKERADDR_ENV_VAR);
    jobSpecificProperties.add(Settings.SPARK_HISTORY_SERVER_ENV);
    jobSpecificProperties.add(Settings.SPARK_NUMBER_EXECUTORS);
    jobSpecificProperties.add("spark.driver.memory");
    jobSpecificProperties.add("spark.driver.cores");
    jobSpecificProperties.add("spark.executor.memory");
    jobSpecificProperties.add("spark.executor.cores");

    addSystemProperty("spark.driver.memory", Integer.toString(driverMemory)+"m");
    addSystemProperty("spark.driver.cores", Integer.toString(driverCores));
    addSystemProperty("spark.executor.memory", executorMemory);
    addSystemProperty("spark.executor.cores", Integer.toString(executorCores));
    
    //Add local resources to spark environment too
    builder.addCommand(new SparkSetEnvironmentCommand());

    InputStream is = null;

    try {
        is = new FileInputStream("/srv/spark/conf/spark-defaults.conf");
    } catch (FileNotFoundException e) {}
    
    //Set up command
    StringBuilder amargs = new StringBuilder("--class ");
    amargs.append(mainClass);
 
    Properties sparkProperties = new Properties();
    sparkProperties.load(is);
    for(String property : sparkProperties.stringPropertyNames()){
        if(!jobSpecificProperties.contains(property)){
            addSystemProperty(property, sparkProperties.getProperty(property).trim());
        }
    }
    for (String s : sysProps.keySet()) {
      String option = escapeForShell("-D" + s + "=" + sysProps.get(s));
      builder.addJavaOption(option);
    }
    
    is.close();
    
    // 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);
    builder.amQueue(driverQueue);

    //Set app name
    builder.appName(jobName);
{'fraction_non_alphanumeric': 0.11018384131591678, 'fraction_numerical': 0.004596032897919691, 'mean_word_length': 3.515838339705079, 'pattern_counts': {'":': 2, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, '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': '3435340', 'n_tokens_mistral': 5047, 'n_tokens_neox': 4859, 'n_words': 1051}