@Test public void testSparkRunnerCommandFull() { String expected = "spark2-submit --conf spark.metrics.conf=metrics.properties " + "--conf \"spark.driver.extraClassPath=logstash-gelf.jar\" " + "--driver-java-options \"-Dlog4j.configuration=file:log4j.properties\" --queue pipelines --conf spark.default.parallelism=1 " + "--conf spark.executor.memoryOverhead=1 --conf spark.dynamicAllocation.enabled=false " + "--class org.gbif.Test --master yarn " + "--deploy-mode cluster --executor-memory 1G --executor-cores 1 --num-executors 1 --driver-memory 4G java.jar " + "--datasetId=de7ffb5e-c07b-42dc-8a88-f67a4465fe3d --attempt=1 --runner=SparkRunner --inputPath=tmp --targetPath=tmp " + "--metaFileName=interpreted-to-index.yml --hdfsSiteConfig=hdfs.xml --coreSiteConfig=core.xml " + "--esHosts=http: IndexingConfiguration config = new IndexingConfiguration(); config.distributedConfig.jarPath = "java.jar"; config.distributedConfig.mainClass = "org.gbif.Test"; config.sparkConfig.executorMemoryGbMax = 10; config.sparkConfig.executorMemoryGbMin = 1; config.sparkConfig.executorCores = 1; config.sparkConfig.executorNumbersMin = 1; config.sparkConfig.executorNumbersMax = 2; config.sparkConfig.memoryOverhead = 1; config.sparkConfig.driverMemory = "4G"; config.distributedConfig.metricsPropertiesPath = "metrics.properties"; config.distributedConfig.extraClassPath = "logstash-gelf.jar"; config.distributedConfig.driverJavaOptions = "-Dlog4j.configuration=file:log4j.properties"; config.distributedConfig.deployMode = "cluster"; config.processRunner = StepRunner.DISTRIBUTED.name(); config.esConfig.hosts = new String[] {"http: config.distributedConfig.yarnQueue = "pipelines"; config.pipelinesConfig = "/path/ws.config"; config.stepConfig.hdfsSiteConfig = "hdfs.xml"; config.stepConfig.repositoryPath = "tmp"; config.stepConfig.coreSiteConfig = "core.xml"; UUID datasetId = UUID.fromString("de7ffb5e-c07b-42dc-8a88-f67a4465fe3d"); int attempt = 1; Set<String> steps = Collections.singleton(RecordType.ALL.name()); ValidationResult vr = new ValidationResult(); PipelinesInterpretedMessage message = new PipelinesInterpretedMessage( datasetId, attempt, steps, 100L, false, null, EndpointType.DWC_ARCHIVE, vr); String indexName = "occurrence"; ProcessBuilder builder = ProcessRunnerBuilder.builder() .config(config) .message(message) .sparkParallelism(1) .sparkExecutorMemory("1G") .sparkExecutorNumbers(1) .esIndexName(indexName) .build() .get(); String result = builder.command().get(2); assertEquals(expected, result); }
ProcessBuilder get() { if (StepRunner.DISTRIBUTED.name().equals(config.processRunner)) { return buildSpark(); } throw new IllegalArgumentException("Wrong runner type - " + config.processRunner); }
ProcessRunnerBuilder { ProcessBuilder get() { if (StepRunner.DISTRIBUTED.name().equals(config.processRunner)) { return buildSpark(); } throw new IllegalArgumentException("Wrong runner type - " + config.processRunner); } }
ProcessRunnerBuilder { ProcessBuilder get() { if (StepRunner.DISTRIBUTED.name().equals(config.processRunner)) { return buildSpark(); } throw new IllegalArgumentException("Wrong runner type - " + config.processRunner); }  }
ProcessRunnerBuilder { ProcessBuilder get() { if (StepRunner.DISTRIBUTED.name().equals(config.processRunner)) { return buildSpark(); } throw new IllegalArgumentException("Wrong runner type - " + config.processRunner); }  String[] buildOptions(); }
ProcessRunnerBuilder { ProcessBuilder get() { if (StepRunner.DISTRIBUTED.name().equals(config.processRunner)) { return buildSpark(); } throw new IllegalArgumentException("Wrong runner type - " + config.processRunner); }  String[] buildOptions();  }