diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/SparkArgsUtils.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/SparkArgsUtils.java index ade087424a..5cc7bd831a 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/SparkArgsUtils.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/SparkArgsUtils.java @@ -19,6 +19,7 @@ package org.apache.dolphinscheduler.server.utils; import org.apache.dolphinscheduler.common.Constants; import org.apache.dolphinscheduler.common.enums.ProgramType; +import org.apache.dolphinscheduler.common.process.ResourceInfo; import org.apache.dolphinscheduler.common.task.spark.SparkParameters; import org.apache.commons.lang.StringUtils; @@ -53,63 +54,69 @@ public class SparkArgsUtils { args.add(param.getDeployMode()); - if(param.getProgramType() !=null ){ - if(param.getProgramType()!=ProgramType.PYTHON){ - if (StringUtils.isNotEmpty(param.getMainClass())) { - args.add(Constants.MAIN_CLASS); - args.add(param.getMainClass()); - } - } + ProgramType type = param.getProgramType(); + String mainClass = param.getMainClass(); + if(type != null && type != ProgramType.PYTHON && StringUtils.isNotEmpty(mainClass)){ + args.add(Constants.MAIN_CLASS); + args.add(mainClass); } - - if (param.getDriverCores() != 0) { + int driverCores = param.getDriverCores(); + if (driverCores != 0) { args.add(Constants.DRIVER_CORES); - args.add(String.format("%d", param.getDriverCores())); + args.add(String.format("%d", driverCores)); } - if (StringUtils.isNotEmpty(param.getDriverMemory())) { + String driverMemory = param.getDriverMemory(); + if (StringUtils.isNotEmpty(driverMemory)) { args.add(Constants.DRIVER_MEMORY); - args.add(param.getDriverMemory()); + args.add(driverMemory); } - if (param.getNumExecutors() != 0) { + int numExecutors = param.getNumExecutors(); + if (numExecutors != 0) { args.add(Constants.NUM_EXECUTORS); - args.add(String.format("%d", param.getNumExecutors())); + args.add(String.format("%d", numExecutors)); } - if (param.getExecutorCores() != 0) { + int executorCores = param.getExecutorCores(); + if (executorCores != 0) { args.add(Constants.EXECUTOR_CORES); - args.add(String.format("%d", param.getExecutorCores())); + args.add(String.format("%d", executorCores)); } - if (StringUtils.isNotEmpty(param.getExecutorMemory())) { + String executorMemory = param.getExecutorMemory(); + if (StringUtils.isNotEmpty(executorMemory)) { args.add(Constants.EXECUTOR_MEMORY); - args.add(param.getExecutorMemory()); + args.add(executorMemory); } // --files --conf --libjar ... - if (StringUtils.isNotEmpty(param.getOthers())) { - String others = param.getOthers(); - if(!others.contains("--queue")){ - if (StringUtils.isNotEmpty(param.getQueue())) { - args.add(Constants.SPARK_QUEUE); - args.add(param.getQueue()); - } + String others = param.getOthers(); + String queue = param.getQueue(); + if (StringUtils.isNotEmpty(others)) { + + if(!others.contains(Constants.SPARK_QUEUE) && StringUtils.isNotEmpty(queue)){ + args.add(Constants.SPARK_QUEUE); + args.add(queue); } - args.add(param.getOthers()); - }else if (StringUtils.isNotEmpty(param.getQueue())) { + + args.add(others); + + }else if (StringUtils.isNotEmpty(queue)) { args.add(Constants.SPARK_QUEUE); - args.add(param.getQueue()); + args.add(queue); } - if (param.getMainJar() != null) { - args.add(param.getMainJar().getRes()); + ResourceInfo mainJar = param.getMainJar(); + if (mainJar != null) { + args.add(mainJar.getRes()); } - if (StringUtils.isNotEmpty(param.getMainArgs())) { - args.add(param.getMainArgs()); + String mainArgs = param.getMainArgs(); + if (StringUtils.isNotEmpty(mainArgs)) { + args.add(mainArgs); } return args; diff --git a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/SparkArgsUtilsTest.java b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/SparkArgsUtilsTest.java new file mode 100644 index 0000000000..6e55fa731b --- /dev/null +++ b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/SparkArgsUtilsTest.java @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.dolphinscheduler.server.utils; + +import org.apache.dolphinscheduler.common.enums.ProgramType; +import org.apache.dolphinscheduler.common.process.ResourceInfo; +import org.apache.dolphinscheduler.common.task.spark.SparkParameters; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; + +/** + * Test SparkArgsUtils + */ +public class SparkArgsUtilsTest { + + private static final Logger logger = LoggerFactory.getLogger(SparkArgsUtilsTest.class); + + public String mode = "cluster"; + public String mainClass = "com.test"; + public ResourceInfo mainJar = null; + public String mainArgs = "partitions=2"; + public String driverMemory = "2G"; + public String executorMemory = "4G"; + public ProgramType programType = ProgramType.JAVA; + public int driverCores = 2; + public int executorCores = 6; + public String sparkVersion = "SPARK1"; + public int numExecutors = 4; + public String queue = "queue1"; + + + @Before + public void setUp() throws Exception { + + ResourceInfo main = new ResourceInfo(); + main.setRes("testspark-1.0.0-SNAPSHOT.jar"); + mainJar = main; + } + + /** + * Test buildArgs + */ + @Test + public void testBuildArgs() { + + //Define params + SparkParameters param = new SparkParameters(); + param.setDeployMode(mode); + param.setMainClass(mainClass); + param.setDriverCores(driverCores); + param.setDriverMemory(driverMemory); + param.setExecutorCores(executorCores); + param.setExecutorMemory(executorMemory); + param.setMainJar(mainJar); + param.setNumExecutors(numExecutors); + param.setProgramType(programType); + param.setSparkVersion(sparkVersion); + param.setMainArgs(mainArgs); + param.setQueue(queue); + + //Invoke buildArgs + List result = SparkArgsUtils.buildArgs(param); + for (String s : result) { + logger.info(s); + } + + //Expected values and order + assertEquals(result.size(),20); + + assertEquals(result.get(0),"--master"); + assertEquals(result.get(1),"yarn"); + + assertEquals(result.get(2),"--deploy-mode"); + assertEquals(result.get(3),mode); + + assertEquals(result.get(4),"--class"); + assertEquals(result.get(5),mainClass); + + assertEquals(result.get(6),"--driver-cores"); + assertSame(Integer.valueOf(result.get(7)),driverCores); + + assertEquals(result.get(8),"--driver-memory"); + assertEquals(result.get(9),driverMemory); + + assertEquals(result.get(10),"--num-executors"); + assertSame(Integer.valueOf(result.get(11)),numExecutors); + + assertEquals(result.get(12),"--executor-cores"); + assertSame(Integer.valueOf(result.get(13)),executorCores); + + assertEquals(result.get(14),"--executor-memory"); + assertEquals(result.get(15),executorMemory); + + assertEquals(result.get(16),"--queue"); + assertEquals(result.get(17),queue); + assertEquals(result.get(18),mainJar.getRes()); + assertEquals(result.get(19),mainArgs); + + //Others param without --queue + SparkParameters param1 = new SparkParameters(); + param1.setOthers("--files xxx/hive-site.xml"); + param1.setQueue(queue); + result = SparkArgsUtils.buildArgs(param1); + assertEquals(result.size(),7); + } +} \ No newline at end of file diff --git a/pom.xml b/pom.xml index 89e8ae91c7..03543cb658 100644 --- a/pom.xml +++ b/pom.xml @@ -653,6 +653,7 @@ **/alert/utils/FuncUtilsTest.java **/alert/utils/JSONUtilsTest.java **/alert/utils/PropertyUtilsTest.java + **/server/utils/SparkArgsUtilsTest.java