<commit_before>package org.gbif.crawler.pipelines.service.interpret;

import org.gbif.crawler.pipelines.config.InterpreterConfiguration;

import java.io.File;
import java.util.Arrays;
import java.util.Objects;
import java.util.Optional;
import java.util.StringJoiner;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Class to build an instance of ProcessBuilder for direct or spark command
 */
public class ProcessRunnerBuilder {

  public enum RunnerEnum {

    DIRECT("DirectRunner"), SPARK("SparkRunner");

    String name;

    RunnerEnum(String name) {
      this.name = name;
    }

    public String getName() {
      return name;
    }

  }

  private static final Logger LOG = LoggerFactory.getLogger(ProcessRunnerBuilder.class);

  private static final String DELIMITER = " ";

  private String user;
  private RunnerEnum runner;
  private String datasetId;
  private int attempt;
  private Integer sparkParallelism;
  private Integer sparkMemoryOverhead;
  private String sparkExecutorMemory;
  private Integer sparkExecutorCores;
  private Integer sparkExecutorNumbers;
  private String sparkDriverMemory;
  private Integer directParallelism;
  private String directStackSize;
  private String directHeapSize;
  private String jarFullPath;
  private String mainClass;
  private String interpretationTypes;
  private String hdfsSiteConfig;
  private String coreSiteConfig;
  private String targetDirectory;
  private String inputFile;
  private String avroCompressionType;
  private Integer avroSyncInterval;
  private String wsConfig;
  private String redirectErrorFile;
  private String redirectOutputFile;
  private String logConfigPath;

  public ProcessRunnerBuilder user(String user) {
    this.user = user;
    return this;
  }

  public ProcessRunnerBuilder runner(RunnerEnum runner) {
    this.runner = runner;
    return this;
  }

  public ProcessRunnerBuilder sparkParallelism(Integer sparkParallelism) {
    this.sparkParallelism = sparkParallelism;
    return this;
  }

  public ProcessRunnerBuilder sparkMemoryOverhead(Integer sparkMemoryOverhead) {
    this.sparkMemoryOverhead = sparkMemoryOverhead;
    return this;
  }

  public ProcessRunnerBuilder sparkExecutorMemory(String sparkExecutorMemory) {
    this.sparkExecutorMemory = sparkExecutorMemory;
    return this;
  }

  public ProcessRunnerBuilder sparkExecutorCores(Integer sparkExecutorCores) {
    this.sparkExecutorCores = sparkExecutorCores;
    return this;
  }

  public ProcessRunnerBuilder sparkExecutorNumbers(Integer sparkExecutorNumbers) {
    this.sparkExecutorNumbers = sparkExecutorNumbers;
    return this;
  }

  public ProcessRunnerBuilder sparkDriverMemory(String sparkDriverMemory) {
    this.sparkDriverMemory = sparkDriverMemory;
    return this;
  }

  public ProcessRunnerBuilder directParallelism(Integer directParallelism) {
    this.directParallelism = directParallelism;
    return this;
  }

  public ProcessRunnerBuilder directStackSize(String directStackSize) {
    this.directStackSize = directStackSize;
    return this;
  }

  public ProcessRunnerBuilder directHeapSize(String directHeapSize) {
    this.directHeapSize = directHeapSize;
    return this;
  }

  public ProcessRunnerBuilder jarFullPath(String jarFullPath) {
    this.jarFullPath = jarFullPath;
    return this;
  }

  public ProcessRunnerBuilder mainClass(String mainClass) {
    this.mainClass = mainClass;
    return this;
  }

  public ProcessRunnerBuilder datasetId(String datasetId) {
    this.datasetId = datasetId;
    return this;
  }

  public ProcessRunnerBuilder attempt(int attempt) {
    this.attempt = attempt;
    return this;
  }

  public ProcessRunnerBuilder interpretationTypes(String... interpretationTypes) {
    this.interpretationTypes = Arrays.stream(interpretationTypes).collect(Collectors.joining(","));
    return this;
  }

  public ProcessRunnerBuilder redirectErrorFile(String redirectErrorFile) {
    this.redirectErrorFile = redirectErrorFile;
    return this;
  }

  public ProcessRunnerBuilder redirectOutputFile(String redirectOutputFile) {
    this.redirectOutputFile = redirectOutputFile;
    return this;
  }

  public ProcessRunnerBuilder logConfigPath(String logConfigPath) {
    this.logConfigPath = logConfigPath;
    return this;
  }

  public ProcessRunnerBuilder hdfsSiteConfig(String hdfsSiteConfig) {
    this.hdfsSiteConfig = hdfsSiteConfig;
    return this;
  }

  public ProcessRunnerBuilder coreSiteConfig(String coreSiteConfig) {
    this.coreSiteConfig = coreSiteConfig;
    return this;
  }

  public ProcessRunnerBuilder targetDirectory(String targetDirectory) {
    this.targetDirectory = targetDirectory;
    return this;
  }

  public ProcessRunnerBuilder inputFile(String inputFile) {
    this.inputFile = inputFile;
    return this;
  }

  public ProcessRunnerBuilder avroCompressionType(String avroCompressionType) {
    this.avroCompressionType = avroCompressionType;
    return this;
  }

  public ProcessRunnerBuilder avroSyncInterval(Integer avroSyncInterval) {
    this.avroSyncInterval = avroSyncInterval;
    return this;
  }

  public ProcessRunnerBuilder wsConfig(String wsConfig) {
    this.wsConfig = wsConfig;
    return this;
  }

  public static ProcessRunnerBuilder create() {
    return new ProcessRunnerBuilder();
  }

  public static ProcessRunnerBuilder create(InterpreterConfiguration config) {
    return ProcessRunnerBuilder.create()
      .user(config.otherUser)
      .directParallelism(config.directParallelism)
      .directStackSize(config.directStackSize)
      .directHeapSize(config.directHeapSize)
      .sparkParallelism(config.sparkParallelism)
      .sparkMemoryOverhead(config.sparkMemoryOverhead)
      .sparkExecutorMemory(config.sparkExecutorMemory)
      .sparkExecutorCores(config.sparkExecutorCores)
      .sparkExecutorNumbers(config.sparkExecutorNumbers)
      .sparkDriverMemory(config.sparkDriverMemory)
      .jarFullPath(config.jarFullPath)
      .mainClass(config.mainClass)
      .hdfsSiteConfig(config.hdfsSiteConfig)
      .coreSiteConfig(config.coreSiteConfig)
      .targetDirectory(config.targetDirectory)
      .avroCompressionType(config.avroConfig.compressionType)
      .avroSyncInterval(config.avroConfig.syncInterval)
      .logConfigPath(config.logConfigPath)
      .wsConfig(config.wsConfig);
  }

  public ProcessBuilder build() {
    if (RunnerEnum.DIRECT == runner) {
      return buildDirect();
    }
    if (RunnerEnum.SPARK == runner) {
      return buildSpark();
    }
    throw new IllegalArgumentException("Wrong runner type - " + runner);
  }

  /**
   * Builds ProcessBuilder to process direct command
   */
  private ProcessBuilder buildDirect() {
    StringJoiner joiner = new StringJoiner(DELIMITER).add("java")
      .add("-XX:+UseG1GC")
      .add("-Xms" + Objects.requireNonNull(directStackSize))
      .add("-Xmx" + Objects.requireNonNull(directHeapSize))
      .add("-Dlogback.configurationFile=" + Objects.requireNonNull(logConfigPath))
      .add("-cp")
      .add(Objects.requireNonNull(jarFullPath))
      .add(Objects.requireNonNull(mainClass))
      .add("--wsProperties=" + Objects.requireNonNull(wsConfig));

    Optional.ofNullable(directParallelism).ifPresent(x -> joiner.add("--targetParallelism=" + x));

    return buildCommon(joiner);
  }

  /**
   * Builds ProcessBuilder to process spark command
   */
  private ProcessBuilder buildSpark() {
    StringJoiner joiner = new StringJoiner(DELIMITER).add("spark-submit")
      .add("--properties-file " + Objects.requireNonNull(wsConfig))
      .add("--conf spark.default.parallelism=" + Objects.requireNonNull(sparkParallelism))
      .add("--conf spark.yarn.executor.memoryOverhead=" + Objects.requireNonNull(sparkMemoryOverhead))
      .add("--class " + Objects.requireNonNull(mainClass))
      .add("--master yarn")
      .add("--deploy-mode cluster")
      .add("--executor-memory " + Objects.requireNonNull(sparkExecutorMemory))
      .add("--executor-cores " + Objects.requireNonNull(sparkExecutorCores))
      .add("--num-executors " + Objects.requireNonNull(sparkExecutorNumbers))
      .add("--driver-memory " + Objects.requireNonNull(sparkDriverMemory))
      .add(Objects.requireNonNull(jarFullPath))
      .add("--wsProperties=" + new File(Objects.requireNonNull(wsConfig)).getName());

    return buildCommon(joiner);
  }

  /**
   * Adds common properties to direct or spark process, for running Java pipelines with pipeline options<commit_msg>pipelines: no need to load the ws.properties in spark anymore<commit_after>package org.gbif.crawler.pipelines.service.interpret;

import org.gbif.crawler.pipelines.config.InterpreterConfiguration;

import java.io.File;
import java.util.Arrays;
import java.util.Objects;
import java.util.Optional;
import java.util.StringJoiner;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Class to build an instance of ProcessBuilder for direct or spark command
 */
public class ProcessRunnerBuilder {

  public enum RunnerEnum {

    DIRECT("DirectRunner"), SPARK("SparkRunner");

    String name;

    RunnerEnum(String name) {
      this.name = name;
    }

    public String getName() {
      return name;
    }

  }

  private static final Logger LOG = LoggerFactory.getLogger(ProcessRunnerBuilder.class);

  private static final String DELIMITER = " ";

  private String user;
  private RunnerEnum runner;
  private String datasetId;
  private int attempt;
  private Integer sparkParallelism;
  private Integer sparkMemoryOverhead;
  private String sparkExecutorMemory;
  private Integer sparkExecutorCores;
  private Integer sparkExecutorNumbers;
  private String sparkDriverMemory;
  private Integer directParallelism;
  private String directStackSize;
  private String directHeapSize;
  private String jarFullPath;
  private String mainClass;
  private String interpretationTypes;
  private String hdfsSiteConfig;
  private String coreSiteConfig;
  private String targetDirectory;
  private String inputFile;
  private String avroCompressionType;
  private Integer avroSyncInterval;
  private String wsConfig;
  private String redirectErrorFile;
  private String redirectOutputFile;
  private String logConfigPath;

  public ProcessRunnerBuilder user(String user) {
    this.user = user;
    return this;
  }

  public ProcessRunnerBuilder runner(RunnerEnum runner) {
    this.runner = runner;
    return this;
  }

  public ProcessRunnerBuilder sparkParallelism(Integer sparkParallelism) {
    this.sparkParallelism = sparkParallelism;
    return this;
  }

  public ProcessRunnerBuilder sparkMemoryOverhead(Integer sparkMemoryOverhead) {
    this.sparkMemoryOverhead = sparkMemoryOverhead;
    return this;
  }

  public ProcessRunnerBuilder sparkExecutorMemory(String sparkExecutorMemory) {
    this.sparkExecutorMemory = sparkExecutorMemory;
    return this;
  }

  public ProcessRunnerBuilder sparkExecutorCores(Integer sparkExecutorCores) {
    this.sparkExecutorCores = sparkExecutorCores;
    return this;
  }

  public ProcessRunnerBuilder sparkExecutorNumbers(Integer sparkExecutorNumbers) {
    this.sparkExecutorNumbers = sparkExecutorNumbers;
    return this;
  }

  public ProcessRunnerBuilder sparkDriverMemory(String sparkDriverMemory) {
    this.sparkDriverMemory = sparkDriverMemory;
    return this;
  }

  public ProcessRunnerBuilder directParallelism(Integer directParallelism) {
    this.directParallelism = directParallelism;
    return this;
  }

  public ProcessRunnerBuilder directStackSize(String directStackSize) {
    this.directStackSize = directStackSize;
    return this;
  }

  public ProcessRunnerBuilder directHeapSize(String directHeapSize) {
    this.directHeapSize = directHeapSize;
    return this;
  }

  public ProcessRunnerBuilder jarFullPath(String jarFullPath) {
    this.jarFullPath = jarFullPath;
    return this;
  }

  public ProcessRunnerBuilder mainClass(String mainClass) {
    this.mainClass = mainClass;
    return this;
  }

  public ProcessRunnerBuilder datasetId(String datasetId) {
    this.datasetId = datasetId;
    return this;
  }

  public ProcessRunnerBuilder attempt(int attempt) {
    this.attempt = attempt;
    return this;
  }

  public ProcessRunnerBuilder interpretationTypes(String... interpretationTypes) {
    this.interpretationTypes = Arrays.stream(interpretationTypes).collect(Collectors.joining(","));
    return this;
  }

  public ProcessRunnerBuilder redirectErrorFile(String redirectErrorFile) {
    this.redirectErrorFile = redirectErrorFile;
    return this;
  }

  public ProcessRunnerBuilder redirectOutputFile(String redirectOutputFile) {
    this.redirectOutputFile = redirectOutputFile;
    return this;
  }

  public ProcessRunnerBuilder logConfigPath(String logConfigPath) {
    this.logConfigPath = logConfigPath;
    return this;
  }

  public ProcessRunnerBuilder hdfsSiteConfig(String hdfsSiteConfig) {
    this.hdfsSiteConfig = hdfsSiteConfig;
    return this;
  }

  public ProcessRunnerBuilder coreSiteConfig(String coreSiteConfig) {
    this.coreSiteConfig = coreSiteConfig;
    return this;
  }

  public ProcessRunnerBuilder targetDirectory(String targetDirectory) {
    this.targetDirectory = targetDirectory;
    return this;
  }

  public ProcessRunnerBuilder inputFile(String inputFile) {
    this.inputFile = inputFile;
    return this;
  }

  public ProcessRunnerBuilder avroCompressionType(String avroCompressionType) {
    this.avroCompressionType = avroCompressionType;
    return this;
  }

  public ProcessRunnerBuilder avroSyncInterval(Integer avroSyncInterval) {
    this.avroSyncInterval = avroSyncInterval;
    return this;
  }

  public ProcessRunnerBuilder wsConfig(String wsConfig) {
    this.wsConfig = wsConfig;
    return this;
  }

  public static ProcessRunnerBuilder create() {
    return new ProcessRunnerBuilder();
  }

  public static ProcessRunnerBuilder create(InterpreterConfiguration config) {
    return ProcessRunnerBuilder.create()
      .user(config.otherUser)
      .directParallelism(config.directParallelism)
      .directStackSize(config.directStackSize)
      .directHeapSize(config.directHeapSize)
      .sparkParallelism(config.sparkParallelism)
      .sparkMemoryOverhead(config.sparkMemoryOverhead)
      .sparkExecutorMemory(config.sparkExecutorMemory)
      .sparkExecutorCores(config.sparkExecutorCores)
      .sparkExecutorNumbers(config.sparkExecutorNumbers)
      .sparkDriverMemory(config.sparkDriverMemory)
      .jarFullPath(config.jarFullPath)
      .mainClass(config.mainClass)
      .hdfsSiteConfig(config.hdfsSiteConfig)
      .coreSiteConfig(config.coreSiteConfig)
      .targetDirectory(config.targetDirectory)
      .avroCompressionType(config.avroConfig.compressionType)
      .avroSyncInterval(config.avroConfig.syncInterval)
      .logConfigPath(config.logConfigPath)
      .wsConfig(config.wsConfig);
  }

  public ProcessBuilder build() {
    if (RunnerEnum.DIRECT == runner) {
      return buildDirect();
    }
    if (RunnerEnum.SPARK == runner) {
      return buildSpark();
    }
    throw new IllegalArgumentException("Wrong runner type - " + runner);
  }

  /**
   * Builds ProcessBuilder to process direct command
   */
  private ProcessBuilder buildDirect() {
    StringJoiner joiner = new StringJoiner(DELIMITER).add("java")
      .add("-XX:+UseG1GC")
      .add("-Xms" + Objects.requireNonNull(directStackSize))
      .add("-Xmx" + Objects.requireNonNull(directHeapSize))
      .add("-Dlogback.configurationFile=" + Objects.requireNonNull(logConfigPath))
      .add("-cp")
      .add(Objects.requireNonNull(jarFullPath))
      .add(Objects.requireNonNull(mainClass))
      .add("--wsProperties=" + Objects.requireNonNull(wsConfig));

    Optional.ofNullable(directParallelism).ifPresent(x -> joiner.add("--targetParallelism=" + x));

    return buildCommon(joiner);
  }

  /**
   * Builds ProcessBuilder to process spark command
   */
  private ProcessBuilder buildSpark() {
    StringJoiner joiner = new StringJoiner(DELIMITER).add("spark-submit")
      .add("--conf spark.default.parallelism=" + Objects.requireNonNull(sparkParallelism))
      .add("--conf spark.yarn.executor.memoryOverhead=" + Objects.requireNonNull(sparkMemoryOverhead))
      .add("--class " + Objects.requireNonNull(mainClass))
      .add("--master yarn")
      .add("--deploy-mode cluster")
      .add("--executor-memory " + Objects.requireNonNull(sparkExecutorMemory))
      .add("--executor-cores " + Objects.requireNonNull(sparkExecutorCores))
      .add("--num-executors " + Objects.requireNonNull(sparkExecutorNumbers))
      .add("--driver-memory " + Objects.requireNonNull(sparkDriverMemory))
      .add(Objects.requireNonNull(jarFullPath))
      .add("--wsProperties=" + Objects.requireNonNull(wsConfig));

    return buildCommon(joiner);
  }

  /**
   * Adds common properties to direct or spark process, for running Java pipelines with pipeline options
{'fraction_non_alphanumeric': 0.08530833628109892, 'fraction_numerical': 0.00035373187124159886, 'mean_word_length': 4.241965389369592, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '3233741', 'n_tokens_mistral': 4774, 'n_tokens_neox': 4662, 'n_words': 977}