test/DolphinScheduler
1
0
Fork 0
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git synced 2024-12-03 20:58:11 +08:00
Code Issues Actions 10 Packages Projects Releases Wiki Activity

Python task specifies path execution

Browse Source
This commit is contained in:
qiaozhanwei 2019-04-16 15:11:18 +08:00
parent dd6a103f58
commit df5d9a583b
2 changed files with 117 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
escheduler-server/src/main/java/cn/escheduler/server/worker/task/PythonCommandExecutor.java
View File

@ -19,9 +19,9 @@ package cn.escheduler.server.worker.task;
import cn.escheduler.common.utils.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.io.*;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
@ -34,6 +34,8 @@ import java.util.function.Consumer;
*/
public class PythonCommandExecutor extends AbstractCommandExecutor {
private static final Logger logger = LoggerFactory.getLogger(PythonCommandExecutor.class);
public static final String PYTHON = "python";
@ -63,27 +65,13 @@ public class PythonCommandExecutor extends AbstractCommandExecutor {
*/
@Override
protected void createCommandFileIfNotExists(String execCommand, String commandFile) throws IOException {
logger.info("proxy user:{}, work dir:{}", tenantCode, taskDir);
logger.info("tenant :{}, work dir:{}", tenantCode, taskDir);
if (!Files.exists(Paths.get(commandFile))) {
logger.info("generate command file:{}", commandFile);
StringBuilder sb = new StringBuilder(200);
sb.append("#-*- encoding=utf8 -*-\n");
sb.append("import os,sys\n");
sb.append("BASEDIR = os.path.dirname(os.path.realpath(__file__))\n");
sb.append("os.chdir(BASEDIR)\n");
if (StringUtils.isNotEmpty(envFile)) {
String[] envArray = envFile.split("\\.");
if(envArray.length == 2){
String path = envArray[0];
logger.info("path:"+path);
int index = path.lastIndexOf("/");
sb.append(String.format("sys.path.append('%s')\n",path.substring(0,index)));
sb.append(String.format("import %s\n",path.substring(index+1)));
}
}
sb.append("\n\n");
sb.append(String.format("import py_%s_node\n",taskAppId));
@ -96,7 +84,12 @@ public class PythonCommandExecutor extends AbstractCommandExecutor {
@Override
protected String commandType() {
return PYTHON;
String envPath = System.getProperty("user.dir")+"/conf/env/.escheduler_env.sh";
String pythonHome = getPythonHome(envPath);
if (StringUtils.isEmpty(pythonHome)){
return PYTHON;
}
return pythonHome;
}
@Override
@ -109,4 +102,45 @@ public class PythonCommandExecutor extends AbstractCommandExecutor {
return true;
}
/**
* get python home
* @param envPath
* @return
*/
private static String getPythonHome(String envPath){
BufferedReader br = null;
String line = null;
StringBuilder sb = new StringBuilder();
try {
br = new BufferedReader(new InputStreamReader(new FileInputStream(envPath)));
while ((line = br.readLine()) != null){
if (line.contains("PYTHON_HOME")){
sb.append(line);
break;
}
}
String result = sb.toString();
if (org.apache.commons.lang.StringUtils.isEmpty(result)){
return null;
}
String[] arrs = result.split("=");
if (arrs.length == 2){
return arrs[1];
}
}catch (IOException e){
logger.error("read file failed : " + e.getMessage(),e);
}finally {
try {
if (br != null){
br.close();
}
} catch (IOException e) {
logger.error(e.getMessage(),e);
}
}
return null;
}
}

65
escheduler-server/src/test/java/cn/escheduler/server/worker/EnvFileTest.java Normal file
View File

@ -0,0 +1,65 @@
package cn.escheduler.server.worker;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
/**
* Created by qiaozhanwei on 2019/4/15.
*/
public class EnvFileTest {
private static final Logger logger = LoggerFactory.getLogger(EnvFileTest.class);
public static void main(String[] args) {
String path = System.getProperty("user.dir")+"\\script\\env\\.escheduler_env.sh";
String pythonHome = getPythonHome(path);
logger.info(pythonHome);
}
/**
* get python home
* @param path
* @return
*/
private static String getPythonHome(String path){
BufferedReader br = null;
String line = null;
StringBuilder sb = new StringBuilder();
try {
br = new BufferedReader(new InputStreamReader(new FileInputStream(path)));
while ((line = br.readLine()) != null){
if (line.contains("PYTHON_HOME")){
sb.append(line);
break;
}
}
String result = sb.toString();
if (StringUtils.isEmpty(result)){
return null;
}
String[] arrs = result.split("=");
if (arrs.length == 2){
return arrs[1];
}
}catch (IOException e){
logger.error("read file failed : " + e.getMessage(),e);
}finally {
try {
if (br != null){
br.close();
}
} catch (IOException e) {
logger.error(e.getMessage(),e);
}
}
return null;
}
}