natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add SparkArgsUtilsTest which is the UT of SparkArgsUtils (#1534) 

1.Add SparkArgsUtilsTest
2.Optimize SparkArgsUtils
3.Add the UT path in pom
This commit is contained in:
zhukai 2019-12-23 15:35:50 +08:00 committed by lgcareer
parent 15a402a3f2
commit d44d06489a
3 changed files with 168 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/SparkArgsUtils.java
View File

@ -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;

128
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/SparkArgsUtilsTest.java Normal file
View File

@ -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<String> 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);
}
}

1
pom.xml
View File

@ -653,6 +653,7 @@
<include>**/alert/utils/FuncUtilsTest.java</include>
<include>**/alert/utils/JSONUtilsTest.java</include>
<include>**/alert/utils/PropertyUtilsTest.java</include>
<include>**/server/utils/SparkArgsUtilsTest.java</include>
</includes>
<!-- <skip>true</skip> -->
</configuration>