This is an automated email from the ASF dual-hosted git repository. aiceflower pushed a commit to branch release-0.9.4 in repository https://gitbox.apache.org/repos/asf/linkis.git
commit 3ebb7011188721cfa526464ca71085bed2f42590 Author: jftang <[email protected]> AuthorDate: Wed Jun 3 18:14:17 2020 +0800 1.add indexes for table 'linkis_task' 2.add field 'engine_start_time' for table 'linkis_task' 3.write the execution script to the hdfs，When the script is too long --- .../linkis/protocol/query/RequestPersistTask.java | 144 ++++++++++++--------- db/linkis_ddl.sql | 9 +- db/module/linkis_query.sql | 7 +- publicService/jobhistory/pom.xml | 6 + .../linkis/jobhistory/dao/TaskMapper.java | 4 +- .../linkis/jobhistory/dao/impl/taskMapper.xml | 9 +- .../linkis/jobhistory/entity/QueryTask.java | 8 ++ .../linkis/jobhistory/entity/QueryTaskVO.java | 31 +++-- .../jobhistory/restful/api/QueryRestfulApi.java | 13 +- .../transitional/TransitionalQueryService.java | 71 ---------- .../jobhistory/conversions/TaskConversions.scala | 40 ++++-- .../linkis/jobhistory/service/QueryService.java | 4 +- .../jobhistory/service/impl/QueryServiceImpl.scala | 81 +++++++++--- .../linkis/jobhistory/util/QueryUtil.scala | 35 ----- .../linkis/jobhistory/util/QueryUtils.scala | 108 ++++++++++++++++ 15 files changed, 343 insertions(+), 227 deletions(-) diff --git a/core/cloudProtocol/src/main/java/com/webank/wedatasphere/linkis/protocol/query/RequestPersistTask.java b/core/cloudProtocol/src/main/java/com/webank/wedatasphere/linkis/protocol/query/RequestPersistTask.java index 0db2c8268b..8bf25b1fb2 100644 --- a/core/cloudProtocol/src/main/java/com/webank/wedatasphere/linkis/protocol/query/RequestPersistTask.java +++ b/core/cloudProtocol/src/main/java/com/webank/wedatasphere/linkis/protocol/query/RequestPersistTask.java @@ -26,17 +26,15 @@ import java.util.Map; /** * Created by enjoyyin on 2018/9/30. */ -public class RequestPersistTask implements Task, QueryProtocol { +public class RequestPersistTask implements Task { private Long taskID; /** - * Instance is an instance of the unified entry where the task is located. ip + port * instance 是指该task所在的统一入口的实例 ip + port */ private String instance; private String execId; private String umUser; /** - * EngineInstance is the instance information of the engine that the task executes, ip+port * engineInstance 是指task执行所请求的engine的实例信息，ip+port */ private String engineInstance; @@ -47,49 +45,72 @@ public class RequestPersistTask implements Task, QueryProtocol { private String status; private Date createdTime; private Date updatedTime; - private String engineType; private Integer errCode; private String errDesc; + private String taskResource; /** - * The executeApplicationName parameter refers to the service the user is looking for, such as spark python R, etc. * executeApplicationName 参数指的是用户所寻求的服务，比如spark python R等等 */ private String executeApplicationName; /** - * requestApplicationName is the name of the creator, such as IDE or WTSS * requestApplicationName 是creator的传参名，例如IDE或WTSS等 */ private String requestApplicationName; /** - * The user adopts the way of executing the script, and the scriptPath is the storage address of the script. - * 用户采用传入执行脚本的方式，scriptPath就是脚本的存储地址 + * source 存放脚本来源，scriptPath是其中一个参数用户采用传入执行脚本的方式，scriptPath就是脚本的存储地址 */ - private String scriptPath; - private java.util.Map<String, String> source; + private Map<String, String> source; /** - * The runType needs to be used in conjunction with the executeApplicationName. If the user selects Spark as the service, he also needs to specify which execution mode to use, such as pySpark RSpark. - * runType and runType are the same attribute, in order to be compatible with the previous code * runType需要和executeApplicationName结合使用，如用户选择了Spark做为服务，他还需要指明使用哪种执行方式，比如pySpark RSpark等 * runType和runType是同一个属性，为了兼容以前的代码 */ private String runType; - private java.util.Map<String, Object> params; + private String engineType; + private Map<String, Object> params; + + private Date engineStartTime; + + public String getEngineType() { + return engineType; + } + public void setEngineType(String engineType) { + this.engineType = engineType; + this.runType = engineType; + } + public Date getEngineStartTime() { + return engineStartTime; + } + public void setEngineStartTime(Date engineStartTime) { + this.engineStartTime = engineStartTime; + } + + public String getRunType() { + return runType; + } + + public void setRunType(String runType) { + this.runType = runType; + this.engineType = runType; + } @Override public String getInstance() { return instance; } + @Override public String getExecId() { return execId; } + @Override public void setInstance(String instance) { this.instance = instance; } + @Override public void setExecId(String execId) { this.execId = execId; @@ -119,14 +140,6 @@ public class RequestPersistTask implements Task, QueryProtocol { this.requestApplicationName = requestApplicationName; } - public String getScriptPath() { - return scriptPath; - } - - public void setScriptPath(String scriptPath) { - this.scriptPath = scriptPath; - } - public Map<String, String> getSource() { return source; } @@ -135,15 +148,6 @@ public class RequestPersistTask implements Task, QueryProtocol { this.source = source; } - public String getRunType() { - return runType; - } - - public void setRunType(String runType) { - this.runType = runType; - this.engineType = runType; - } - public Long getTaskID() { return taskID; } @@ -152,7 +156,6 @@ public class RequestPersistTask implements Task, QueryProtocol { this.taskID = taskID; } - public String getUmUser() { return umUser; } @@ -161,8 +164,6 @@ public class RequestPersistTask implements Task, QueryProtocol { this.umUser = umUser; } - - public Float getProgress() { return progress; } @@ -203,15 +204,6 @@ public class RequestPersistTask implements Task, QueryProtocol { this.updatedTime = updatedTime; } - public String getEngineType() { - return engineType; - } - - public void setEngineType(String engineType) { - this.engineType = engineType; - this.runType = engineType; - } - public Integer getErrCode() { return errCode; } @@ -232,7 +224,7 @@ public class RequestPersistTask implements Task, QueryProtocol { return executionCode; } - public String getCode(){ + public String getCode() { return this.getExecutionCode(); } @@ -256,6 +248,14 @@ public class RequestPersistTask implements Task, QueryProtocol { this.engineInstance = engineInstance; } + public String getTaskResource() { + return taskResource; + } + + public void setTaskResource(String taskResource) { + this.taskResource = taskResource; + } + @Override public String toString() { return "RequestPersistTask{" + @@ -271,36 +271,48 @@ public class RequestPersistTask implements Task, QueryProtocol { ", status='" + status + '\'' + ", createdTime=" + createdTime + ", updatedTime=" + updatedTime + - ", engineType='" + engineType + '\'' + ", errCode=" + errCode + ", errDesc='" + errDesc + '\'' + + ", executeApplicationName='" + executeApplicationName + '\'' + + ", requestApplicationName='" + requestApplicationName + '\'' + + ", source=" + source + + ", runType='" + runType + '\'' + + ", params=" + params + '}'; } @Override public boolean equals(Object o) { - if (this == o) return true; + if (this == o) { + return true; + } - if (o == null || getClass() != o.getClass()) return false; + if (o == null || getClass() != o.getClass()) { + return false; + } - RequestPersistTask that = (RequestPersistTask) o; + RequestPersistTask task = (RequestPersistTask) o; return new EqualsBuilder() - .append(taskID, that.taskID) - .append(instance, that.instance) - .append(execId, that.execId) - .append(umUser, that.umUser) - .append(engineInstance, that.engineInstance) - .append(executionCode, that.executionCode) - .append(progress, that.progress) - .append(logPath, that.logPath) - .append(resultLocation, that.resultLocation) - .append(status, that.status) - .append(createdTime, that.createdTime) - .append(updatedTime, that.updatedTime) - .append(engineType, that.engineType) - .append(errCode, that.errCode) - .append(errDesc, that.errDesc) + .append(taskID, task.taskID) + .append(instance, task.instance) + .append(execId, task.execId) + .append(umUser, task.umUser) + .append(engineInstance, task.engineInstance) + .append(executionCode, task.executionCode) + .append(progress, task.progress) + .append(logPath, task.logPath) + .append(resultLocation, task.resultLocation) + .append(status, task.status) + .append(createdTime, task.createdTime) + .append(updatedTime, task.updatedTime) + .append(errCode, task.errCode) + .append(errDesc, task.errDesc) + .append(executeApplicationName, task.executeApplicationName) + .append(requestApplicationName, task.requestApplicationName) + .append(source, task.source) + .append(runType, task.runType) + .append(params, task.params) .isEquals(); } @@ -319,9 +331,13 @@ public class RequestPersistTask implements Task, QueryProtocol { .append(status) .append(createdTime) .append(updatedTime) - .append(engineType) .append(errCode) .append(errDesc) + .append(executeApplicationName) + .append(requestApplicationName) + .append(source) + .append(runType) + .append(params) .toHashCode(); } -} +} \ No newline at end of file diff --git a/db/linkis_ddl.sql b/db/linkis_ddl.sql index b8462fb409..8a6795f8e9 100644 --- a/db/linkis_ddl.sql +++ b/db/linkis_ddl.sql @@ -176,7 +176,7 @@ CREATE TABLE `linkis_task` ( `instance` varchar(50) DEFAULT NULL COMMENT 'An instance of Entrance, consists of IP address of the entrance server and port', `exec_id` varchar(50) DEFAULT NULL COMMENT 'execution ID, consists of jobID(generated by scheduler), executeApplicationName , creator and instance', `um_user` varchar(50) DEFAULT NULL COMMENT 'User name', - `execution_code` text, + `execution_code` text COMMENT 'Run script. When exceeding 6000 lines, script would be stored in HDFS and its file path would be stored in database', `progress` float DEFAULT NULL COMMENT 'Script execution progress, between zero and one', `log_path` varchar(200) DEFAULT NULL COMMENT 'File path of the log files', `result_location` varchar(200) DEFAULT NULL COMMENT 'File path of the result', @@ -191,8 +191,11 @@ CREATE TABLE `linkis_task` ( `script_path` varchar(200) DEFAULT NULL COMMENT 'Path of the script in workspace', `params` text COMMENT 'Configuration item of the parameters', `engine_instance` varchar(50) DEFAULT NULL COMMENT 'An instance of engine, consists of IP address of the engine server and port', - PRIMARY KEY (`id`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + `engine_start_time` time DEFAULT NULL, + PRIMARY KEY (`id`), + KEY `created_time` (`created_time`), + KEY `um_user` (`um_user`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; DROP TABLE IF EXISTS `linkis_em_resource_meta_data`; diff --git a/db/module/linkis_query.sql b/db/module/linkis_query.sql index aaef124fef..1262fba7b8 100644 --- a/db/module/linkis_query.sql +++ b/db/module/linkis_query.sql @@ -24,7 +24,10 @@ CREATE TABLE `linkis_task` ( `script_path` varchar(200) DEFAULT NULL COMMENT 'Path of the script in workspace', `params` text COMMENT 'Configuration item of the parameters', `engine_instance` varchar(50) DEFAULT NULL COMMENT 'An instance of engine, consists of IP address of the engine server and port', - PRIMARY KEY (`id`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + `engine_start_time` time DEFAULT NULL, + PRIMARY KEY (`id`), + KEY `created_time` (`created_time`), + KEY `um_user` (`um_user`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/publicService/jobhistory/pom.xml b/publicService/jobhistory/pom.xml index 43421fcaa1..182ebfd3a0 100644 --- a/publicService/jobhistory/pom.xml +++ b/publicService/jobhistory/pom.xml @@ -41,6 +41,12 @@ <version>${linkis.version}</version> <scope>provided</scope> </dependency> + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-storage</artifactId> + <version>${linkis.version}</version> + <scope>provided</scope> + </dependency> <!--<dependency> <groupId>com.github.pagehelper</groupId> <artifactId>pagehelper-spring-boot-starter</artifactId> diff --git a/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/dao/TaskMapper.java b/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/dao/TaskMapper.java index 7d831244dd..338450cf81 100644 --- a/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/dao/TaskMapper.java +++ b/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/dao/TaskMapper.java @@ -33,8 +33,8 @@ public interface TaskMapper { void updateTask(QueryTask queryTask); - List<QueryTask> search(@Param("taskID")Long taskID, @Param("umUser")String username, @Param("status")List<String> status, - @Param("startDate")Date startDate, @Param("endDate")Date endDate,@Param("executeApplicationName")String executeApplicationName); + List<QueryTask> search(@Param("taskID") Long taskID, @Param("umUser") String username, @Param("status") List<String> status, + @Param("startDate") Date startDate, @Param("endDate") Date endDate, @Param("executeApplicationName") String executeApplicationName); String selectTaskStatusForUpdate(Long taskID); } diff --git a/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/dao/impl/taskMapper.xml b/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/dao/impl/taskMapper.xml index a5bc561552..d06ba77dad 100644 --- a/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/dao/impl/taskMapper.xml +++ b/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/dao/impl/taskMapper.xml @@ -28,7 +28,8 @@ <sql id="task_List"> id, `instance`, `exec_id`, `um_user`, `execution_code`, `progress`,`log_path`, - `result_location`,`status`,`created_time`,`updated_time`,`run_type`,`err_code`,`err_desc`,`execute_application_name`,`request_application_name`,`script_path`,`params`,`engine_instance` + `result_location`,`status`,`created_time`,`updated_time`,`run_type`,`err_code`,`err_desc`,`execute_application_name`,`request_application_name`,`script_path`,`params`,`engine_instance`, + `engine_start_time` </sql> <insert id="insertTask" useGeneratedKeys="true" keyProperty="taskID" parameterType="com.webank.wedatasphere.linkis.jobhistory.entity.QueryTask"> @@ -37,7 +38,9 @@ #{umUser},#{executionCode},#{progress}, #{logPath},#{resultLocation},#{status}, #{createdTime},#{updatedTime},#{runType}, - #{errCode},#{errDesc},#{executeApplicationName},#{requestApplicationName},#{sourceJson},#{paramsJson},#{engineInstance}) + #{errCode},#{errDesc},#{executeApplicationName}, + #{requestApplicationName},#{sourceJson},#{paramsJson}, + #{engineInstance},#{engineStartTime}) </insert> <select id="selectTask" resultMap="taskMapper" parameterType="com.webank.wedatasphere.linkis.jobhistory.entity.QueryTask"> @@ -62,6 +65,7 @@ <if test="requestApplicationName != null">and request_application_name = #{requestApplicationName}</if> <if test="sourceJson != null">and script_path = #{sourceJson}</if> <if test="paramsJson != null">and params = #{paramsJson}</if> + <if test="engineStartTime != null">and engine_start_time = #{engineStartTime}</if> </where> </select> @@ -99,6 +103,7 @@ <if test="sourceJson != null"> script_path = #{sourceJson},</if> <if test="paramsJson != null"> params = #{paramsJson},</if> <if test="engineInstance != null"> engine_instance = #{engineInstance},</if> + <if test="engineStartTime != null"> engine_start_time = #{engineStartTime},</if> </trim> WHERE id =#{taskID} </update> diff --git a/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/entity/QueryTask.java b/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/entity/QueryTask.java index 6dee1740bc..c989eb817c 100644 --- a/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/entity/QueryTask.java +++ b/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/entity/QueryTask.java @@ -42,6 +42,14 @@ public class QueryTask { private String runType; private String paramsJson; private String sourceJson; + private Date engineStartTime; + public Date getEngineStartTime() { + return engineStartTime; + } + + public void setEngineStartTime(Date engineStartTime) { + this.engineStartTime = engineStartTime; + } public String getSourceJson() { return sourceJson; diff --git a/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/entity/QueryTaskVO.java b/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/entity/QueryTaskVO.java index 2428f33ad4..50d53189b3 100644 --- a/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/entity/QueryTaskVO.java +++ b/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/entity/QueryTaskVO.java @@ -39,12 +39,33 @@ public class QueryTaskVO { private String errDesc; private String executeApplicationName; private String requestApplicationName; - private String scriptPath; private String runType; private String paramsJson; private Long costTime; private String strongerExecId; private String sourceJson; + /** + * source字段:用来将sourceJson的value取出来进行拼接返回给前台展示 + */ + private String sourceTailor; + + private Date engineStartTime; + + public Date getEngineStartTime() { + return engineStartTime; + } + + public void setEngineStartTime(Date engineStartTime) { + this.engineStartTime = engineStartTime; + } + + public String getSourceTailor() { + return sourceTailor; + } + + public void setSourceTailor(String sourceTailor) { + this.sourceTailor = sourceTailor; + } public String getSourceJson() { return sourceJson; @@ -190,14 +211,6 @@ public class QueryTaskVO { this.requestApplicationName = requestApplicationName; } - public String getScriptPath() { - return scriptPath; - } - - public void setScriptPath(String scriptPath) { - this.scriptPath = scriptPath; - } - public String getRunType() { return runType; } diff --git a/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/restful/api/QueryRestfulApi.java b/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/restful/api/QueryRestfulApi.java index 4be67b75c3..41e1af8332 100644 --- a/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/restful/api/QueryRestfulApi.java +++ b/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/restful/api/QueryRestfulApi.java @@ -22,7 +22,6 @@ import com.webank.wedatasphere.linkis.jobhistory.entity.QueryTask; import com.webank.wedatasphere.linkis.jobhistory.entity.QueryTaskVO; import com.webank.wedatasphere.linkis.jobhistory.exception.QueryException; import com.webank.wedatasphere.linkis.jobhistory.service.QueryService; -import com.webank.wedatasphere.linkis.jobhistory.util.QueryUtil; import com.webank.wedatasphere.linkis.server.Message; import com.webank.wedatasphere.linkis.server.security.SecurityFilter; import org.slf4j.Logger; @@ -80,7 +79,6 @@ public class QueryRestfulApi{ if (startDate != null && endDate == null){ endDate = System.currentTimeMillis(); } - PageHelper.startPage(pageNow,pageSize); Date sDate = null; Date eDate = null; if (startDate != null){ @@ -93,11 +91,18 @@ public class QueryRestfulApi{ instance.add(Calendar.DAY_OF_MONTH,1); eDate = instance.getTime(); } - List<QueryTask> queryTasks = queryService.search(taskID,username,status, sDate,eDate,executeApplicationName); + List<QueryTask> queryTasks = null; + PageHelper.startPage(pageNow,pageSize); + try { + queryTasks = queryService.search(taskID,username,status, sDate,eDate,executeApplicationName); + }finally { + PageHelper.clearPage(); + } + PageInfo<QueryTask> pageInfo = new PageInfo<>(queryTasks); List<QueryTask> list = pageInfo.getList(); long total = pageInfo.getTotal(); - List<QueryTaskVO> vos = QueryUtil.getQueryVOList(list); + List<QueryTaskVO> vos = queryService.getQueryVOList(list); return Message.messageToResponse(Message.ok().data("tasks", vos).data("totalPage",total)); } } diff --git a/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/transitional/TransitionalQueryService.java b/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/transitional/TransitionalQueryService.java deleted file mode 100644 index d86892eb7f..0000000000 --- a/publicService/jobhistory/src/main/java/com/webank/wedatasphere/linkis/jobhistory/transitional/TransitionalQueryService.java +++ /dev/null @@ -1,71 +0,0 @@ -/* - * Copyright 2019 WeBank - * - * Licensed 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 com.webank.wedatasphere.linkis.jobhistory.transitional; - -import com.webank.wedatasphere.linkis.protocol.query.RequestUpdateTask; -import com.webank.wedatasphere.linkis.protocol.query.ResponsePersist; -import com.webank.wedatasphere.linkis.jobhistory.dao.TaskMapper; -import com.webank.wedatasphere.linkis.jobhistory.exception.QueryException; -import com.webank.wedatasphere.linkis.jobhistory.service.impl.QueryServiceImpl; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.stereotype.Component; -import org.springframework.transaction.annotation.Transactional; - -import java.util.HashMap; - -/** - * Created by johnnwang on 2019/6/6. - */ -@Component -public class TransitionalQueryService { - - private Logger logger = LoggerFactory.getLogger(this.getClass()); - - @Autowired - private TaskMapper taskMapper; - - @Autowired - private QueryServiceImpl queryServiceImpl; - - @Transactional - public ResponsePersist change(RequestUpdateTask requestUpdateTask) { - ResponsePersist persist = new ResponsePersist(); - try { - if (requestUpdateTask.getStatus() != null) { - String oldStatus = taskMapper.selectTaskStatusForUpdate(requestUpdateTask.getTaskID()); - if (oldStatus != null && !shouldUpdate(oldStatus, requestUpdateTask.getStatus())) - throw new QueryException(requestUpdateTask.getTaskID() + "The task state in the database is(数据库中的task状态为)：" + oldStatus + "The updated task status is(更新的task状态为)：" + requestUpdateTask.getStatus() + "Update failed!(更新失败)！"); - } - taskMapper.updateTask(queryServiceImpl.requestPersistTaskTask2QueryTask(requestUpdateTask)); - HashMap<String, Object> map = new HashMap<>(); - map.put("taskID", requestUpdateTask.getTaskID()); - persist.setStatus(0); - persist.setData(map); - } catch (Exception e) { - logger.error(e.getMessage()); - persist.setStatus(1); - persist.setMsg(e.getMessage()); - } - return persist; - } - - private boolean shouldUpdate(String oldStatus, String newStatus) { - return TaskStatus.valueOf(oldStatus).ordinal() <= TaskStatus.valueOf(newStatus).ordinal(); - } -} diff --git a/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/conversions/TaskConversions.scala b/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/conversions/TaskConversions.scala index 07a89e61dd..bc370c2db5 100644 --- a/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/conversions/TaskConversions.scala +++ b/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/conversions/TaskConversions.scala @@ -23,6 +23,8 @@ import com.webank.wedatasphere.linkis.common.utils.{Logging, Utils} import com.webank.wedatasphere.linkis.protocol.query.{RequestPersistTask, RequestQueryTask} import com.webank.wedatasphere.linkis.protocol.utils.ZuulEntranceUtils import com.webank.wedatasphere.linkis.jobhistory.entity.{QueryTask, QueryTaskVO} +import com.webank.wedatasphere.linkis.jobhistory.transitional.TaskStatus +import com.webank.wedatasphere.linkis.jobhistory.util.QueryUtils import com.webank.wedatasphere.linkis.protocol.constants.TaskConstant import com.webank.wedatasphere.linkis.server.BDPJettyServerHelper import org.springframework.beans.BeanUtils @@ -33,6 +35,7 @@ import org.springframework.util.StringUtils */ object TaskConversions extends Logging{ + @Deprecated implicit def requestQueryTask2QueryTask(requestQueryTask: RequestQueryTask): QueryTask = { val task: QueryTask = new QueryTask BeanUtils.copyProperties(requestQueryTask, task) @@ -44,6 +47,7 @@ object TaskConversions extends Logging{ } implicit def queryTask2RequestPersistTask(queryTask: QueryTask): RequestPersistTask = { + QueryUtils.exchangeExecutionCode(queryTask) val task = new RequestPersistTask BeanUtils.copyProperties(queryTask, task) task.setSource(BDPJettyServerHelper.gson.fromJson(queryTask.getSourceJson, classOf[java.util.HashMap[String, String]])) @@ -51,6 +55,7 @@ object TaskConversions extends Logging{ task } + @Deprecated implicit def requestPersistTaskTask2QueryTask(requestPersistTask: RequestPersistTask): QueryTask = { val task: QueryTask = new QueryTask BeanUtils.copyProperties(requestPersistTask, task) @@ -62,8 +67,18 @@ object TaskConversions extends Logging{ } implicit def queryTask2QueryTaskVO(queryTask: QueryTask): QueryTaskVO = { + QueryUtils.exchangeExecutionCode(queryTask) val taskVO = new QueryTaskVO BeanUtils.copyProperties(queryTask, taskVO) + if(!StringUtils.isEmpty(taskVO.getSourceJson)){ + Utils.tryCatch{ + val source = BDPJettyServerHelper.gson.fromJson(taskVO.getSourceJson,classOf[util.Map[String,String]]) + import scala.collection.JavaConversions._ + taskVO.setSourceTailor(source.map(_._2).foldLeft("")(_ + _ + "-").stripSuffix("-")) + }{ + case _ =>warn("sourceJson deserializae failed,this task may be the old data") + } + } if (queryTask.getExecId() != null && queryTask.getExecuteApplicationName() != null && queryTask.getInstance() != null) { taskVO.setStrongerExecId(ZuulEntranceUtils.generateExecID(queryTask.getExecId(), queryTask.getExecuteApplicationName(), queryTask.getInstance(), queryTask.getRequestApplicationName)) @@ -71,19 +86,18 @@ object TaskConversions extends Logging{ val status = queryTask.getStatus() val createdTime = queryTask.getCreatedTime() val updatedTime = queryTask.getUpdatedTime() - if ("Succeed".equalsIgnoreCase(status) || "Failed".equalsIgnoreCase(status) || "Cancelled".equalsIgnoreCase(status)) { - if (createdTime != null && updatedTime != null) { - taskVO.setCostTime(queryTask.getUpdatedTime().getTime() - queryTask.getCreatedTime().getTime()); - } else { - taskVO.setCostTime(null) - } - } else { - if (createdTime != null) { - taskVO.setCostTime(System.currentTimeMillis() - queryTask.getCreatedTime().getTime()); - } else { - taskVO.setCostTime(null) - } + if (isJobFinished(status) && createdTime != null && updatedTime != null) { + taskVO.setCostTime(queryTask.getUpdatedTime().getTime() - queryTask.getCreatedTime().getTime()); + } else if (createdTime != null) { + taskVO.setCostTime(System.currentTimeMillis() - queryTask.getCreatedTime().getTime()); } taskVO } -} + + def isJobFinished(status: String): Boolean = { + TaskStatus.Succeed.toString.equals(status) || + TaskStatus.Failed.toString.equals(status) || + TaskStatus.Cancelled.toString.equals(status) || + TaskStatus.Timeout.toString.equals(status) + } +} \ No newline at end of file diff --git a/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/service/QueryService.java b/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/service/QueryService.java index 60e247aece..a4ec6d01f2 100644 --- a/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/service/QueryService.java +++ b/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/service/QueryService.java @@ -34,7 +34,9 @@ public interface QueryService { ResponsePersist query(RequestQueryTask requestQueryTask); - QueryTaskVO getTaskByID(Long taskID,String userName); + QueryTaskVO getTaskByID(Long taskID, String userName); List<QueryTask> search(Long taskID, String username, String status, Date sDate, Date eDate, String executeApplicationName); + + List<QueryTaskVO> getQueryVOList(List<QueryTask> list); } diff --git a/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/service/impl/QueryServiceImpl.scala b/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/service/impl/QueryServiceImpl.scala index ba170d479a..95af6e9d35 100644 --- a/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/service/impl/QueryServiceImpl.scala +++ b/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/service/impl/QueryServiceImpl.scala @@ -16,33 +16,34 @@ package com.webank.wedatasphere.linkis.jobhistory.service.impl -import java.lang.Long import java.util import java.util.Date import com.webank.wedatasphere.linkis.common.utils.{Logging, Utils} import com.webank.wedatasphere.linkis.jobhistory.dao.TaskMapper import com.webank.wedatasphere.linkis.jobhistory.entity.{QueryTask, QueryTaskVO} +import com.webank.wedatasphere.linkis.jobhistory.exception.QueryException import com.webank.wedatasphere.linkis.jobhistory.service.QueryService -import com.webank.wedatasphere.linkis.jobhistory.transitional.TransitionalQueryService +import com.webank.wedatasphere.linkis.jobhistory.transitional.TaskStatus +import com.webank.wedatasphere.linkis.jobhistory.util.QueryUtils import com.webank.wedatasphere.linkis.protocol.query._ import com.webank.wedatasphere.linkis.server.BDPJettyServerHelper import org.springframework.beans.BeanUtils import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service +import org.springframework.transaction.annotation.Transactional /** - * Created by johnnwang on 2019/2/25. - */ + * Created by johnnwang on 2019/2/25. + */ @Service class QueryServiceImpl extends QueryService with Logging { @Autowired private var taskMapper: TaskMapper = _ - @Autowired - private var transitionalQueryService:TransitionalQueryService = _ override def add(requestInsertTask: RequestInsertTask): ResponsePersist = { info("Insert data into the database(往数据库中插入数据)：" + requestInsertTask.toString) + QueryUtils.storeExecutionCode(requestInsertTask) val persist = new ResponsePersist Utils.tryCatch { val queryTask = requestPersistTaskTask2QueryTask(requestInsertTask) @@ -54,18 +55,47 @@ class QueryServiceImpl extends QueryService with Logging { } { case e: Exception => error(e.getMessage) - persist.setStatus(1); - persist.setMsg(e.getMessage); + persist.setStatus(1) + persist.setMsg(e.getMessage) } persist } + @Transactional override def change(requestUpdateTask: RequestUpdateTask): ResponsePersist = { + val executionCode = requestUpdateTask.getExecutionCode + requestUpdateTask.setExecutionCode(null) info("Update data to the database(往数据库中更新数据)：" + requestUpdateTask.toString) - transitionalQueryService.change(requestUpdateTask) + val persist = new ResponsePersist + Utils.tryCatch { + if (requestUpdateTask.getErrDesc != null) { + if (requestUpdateTask.getErrDesc.length > 256) { + info(s"errorDesc is too long,we will cut some message") + requestUpdateTask.setErrDesc(requestUpdateTask.getErrDesc.substring(0, 256)) + info(s"${requestUpdateTask.getErrDesc}") + } + } + if (requestUpdateTask.getStatus != null) { + val oldStatus: String = taskMapper.selectTaskStatusForUpdate(requestUpdateTask.getTaskID) + if (oldStatus != null && !shouldUpdate(oldStatus, requestUpdateTask.getStatus)) + throw new QueryException(s"${requestUpdateTask.getTaskID}数据库中的task状态为：${oldStatus}更新的task状态为：${requestUpdateTask.getStatus}更新失败！") + } + taskMapper.updateTask(requestPersistTaskTask2QueryTask(requestUpdateTask)) + val map = new util.HashMap[String, Object] + map.put("taskID", requestUpdateTask.getTaskID) + persist.setStatus(0) + persist.setData(map) + } { + case e: Exception => + error(e.getMessage) + persist.setStatus(1); + persist.setMsg(e.getMessage); + } + persist } override def query(requestQueryTask: RequestQueryTask): ResponsePersist = { + info("查询历史task：" + requestQueryTask.toString) val persist = new ResponsePersist Utils.tryCatch { val task = taskMapper.selectTask(requestPersistTaskTask2QueryTask(requestQueryTask)) @@ -90,34 +120,43 @@ class QueryServiceImpl extends QueryService with Logging { tasks } - def requestPersistTaskTask2QueryTask(requestPersistTask: RequestPersistTask): QueryTask = { + private def requestPersistTaskTask2QueryTask(requestPersistTask: RequestPersistTask): QueryTask = { val task: QueryTask = new QueryTask BeanUtils.copyProperties(requestPersistTask, task) - if(requestPersistTask.getSource != null) + if (requestPersistTask.getSource != null) task.setSourceJson(BDPJettyServerHelper.gson.toJson(requestPersistTask.getSource)) if (requestPersistTask.getParams != null) task.setParamsJson(BDPJettyServerHelper.gson.toJson(requestPersistTask.getParams)) - else - task.setParamsJson(null) task } - override def getTaskByID(taskID: Long,userName:String): QueryTaskVO = { + override def getTaskByID(taskID: java.lang.Long, userName: String): QueryTaskVO = { val task = new QueryTask task.setTaskID(taskID) task.setUmUser(userName) val taskR = taskMapper.selectTask(task) - if (taskR.size() > 0) { - import com.webank.wedatasphere.linkis.jobhistory.conversions.TaskConversions.queryTask2QueryTaskVO - taskR.get(0) - } else { - null - } + import com.webank.wedatasphere.linkis.jobhistory.conversions.TaskConversions.queryTask2QueryTaskVO + + import scala.collection.JavaConversions._ + if (taskR.isEmpty) null else taskR(0) } - override def search(taskID: Long, username: String, status: String, sDate: Date, eDate: Date, executeApplicationName: String): util.List[QueryTask] = { + override def search(taskID: java.lang.Long, username: String, status: String, sDate: Date, eDate: Date, executeApplicationName: String): util.List[QueryTask] = { import scala.collection.JavaConversions._ val split: util.List[String] = if (status != null) status.split(",").toList else null taskMapper.search(taskID, username, split, sDate, eDate, executeApplicationName) } + + def getQueryVOList(list: java.util.List[QueryTask]): java.util.List[QueryTaskVO] = { + val ovs = new util.ArrayList[QueryTaskVO] + import scala.collection.JavaConversions._ + list.foreach(f => { + import com.webank.wedatasphere.linkis.jobhistory.conversions.TaskConversions.queryTask2QueryTaskVO + ovs.add(f) + }) + ovs + } + + private def shouldUpdate(oldStatus: String, newStatus: String): Boolean = TaskStatus.valueOf(oldStatus).ordinal <= TaskStatus.valueOf(newStatus).ordinal } + diff --git a/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/util/QueryUtil.scala b/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/util/QueryUtil.scala deleted file mode 100644 index c99a22df56..0000000000 --- a/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/util/QueryUtil.scala +++ /dev/null @@ -1,35 +0,0 @@ -/* - * Copyright 2019 WeBank - * - * Licensed 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 com.webank.wedatasphere.linkis.jobhistory.util - -import java.util -import scala.collection.JavaConversions._ -import com.webank.wedatasphere.linkis.jobhistory.entity.{QueryTask, QueryTaskVO} - -/** - * Created by johnnwang on 2019/2/25. - */ -object QueryUtil { - def getQueryVOList(list:java.util.List[QueryTask]):java.util.List[QueryTaskVO] ={ - val ovs = new util.ArrayList[QueryTaskVO] - list.foreach(f =>{ - import com.webank.wedatasphere.linkis.jobhistory.conversions.TaskConversions.queryTask2QueryTaskVO - ovs.add(f) - }) - ovs - } -} diff --git a/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/util/QueryUtils.scala b/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/util/QueryUtils.scala new file mode 100644 index 0000000000..789e4cec38 --- /dev/null +++ b/publicService/jobhistory/src/main/scala/com/webank/wedatasphere/linkis/jobhistory/util/QueryUtils.scala @@ -0,0 +1,108 @@ +/* + * Copyright 2019 WeBank + * + * Licensed 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 com.webank.wedatasphere.linkis.jobhistory.util + +import java.io.{InputStream, OutputStream} +import java.util.Date + +import com.webank.wedatasphere.linkis.common.conf.CommonVars +import com.webank.wedatasphere.linkis.common.io.FsPath +import com.webank.wedatasphere.linkis.common.utils.{Logging, Utils} +import com.webank.wedatasphere.linkis.jobhistory.entity.QueryTask +import com.webank.wedatasphere.linkis.protocol.query.RequestInsertTask +import com.webank.wedatasphere.linkis.storage.FSFactory +import com.webank.wedatasphere.linkis.storage.fs.FileSystem +import com.webank.wedatasphere.linkis.storage.utils.{FileSystemUtils, StorageUtils} +import org.apache.commons.io.IOUtils +import org.apache.commons.lang.time.DateFormatUtils + +/** + * Created by patinousward on 2019/2/25. + */ +object QueryUtils extends Logging { + + private val CODE_STORE_PREFIX = CommonVars("bdp.dataworkcloud.query.store.prefix", "hdfs:///tmp/bdp-ide/") + private val CODE_STORE_SUFFIX = CommonVars("bdp.dataworkcloud.query.store.suffix", "") + private val CHARSET = "utf-8" + private val CODE_SPLIT = ";" + private val LENGTH_SPLIT = "#" + + def storeExecutionCode(requestInsertTask: RequestInsertTask): Unit = { + if (requestInsertTask.getExecutionCode.length < 60000) return + val user: String = requestInsertTask.getUmUser + val path: String = getCodeStorePath(user) + val fsPath: FsPath = new FsPath(path) + val fileSystem = FSFactory.getFsByProxyUser(fsPath, user).asInstanceOf[FileSystem] + fileSystem.init(null) + var os: OutputStream = null + var position = 0L + val codeBytes = requestInsertTask.getExecutionCode.getBytes(CHARSET) + path.intern() synchronized { + Utils.tryFinally { + if (!fileSystem.exists(fsPath)) FileSystemUtils.createNewFile(fsPath, user, true) + os = fileSystem.write(fsPath, false) + position = fileSystem.get(path).getLength + IOUtils.write(codeBytes, os) + } { + IOUtils.closeQuietly(os) + if (fileSystem != null) fileSystem.close() + } + } + val length = codeBytes.length + requestInsertTask.setExecutionCode(path + CODE_SPLIT + position + LENGTH_SPLIT + length) + } + + def exchangeExecutionCode(queryTask: QueryTask): Unit = { + import scala.util.control.Breaks._ + if (queryTask.getExecutionCode == null || !queryTask.getExecutionCode.startsWith(StorageUtils.HDFS_SCHEMA)) return + val codePath = queryTask.getExecutionCode + val path = codePath.substring(0, codePath.lastIndexOf(CODE_SPLIT)) + val codeInfo = codePath.substring(codePath.lastIndexOf(CODE_SPLIT) + 1) + val infos: Array[String] = codeInfo.split(LENGTH_SPLIT) + val position = infos(0).toInt + var lengthLeft = infos(1).toInt + val tub = new Array[Byte](1024) + val executionCode: StringBuilder = new StringBuilder + val fsPath: FsPath = new FsPath(path) + val fileSystem = FSFactory.getFsByProxyUser(fsPath, queryTask.getUmUser).asInstanceOf[FileSystem] + fileSystem.init(null) + var is: InputStream = null + if (!fileSystem.exists(fsPath)) return + Utils.tryFinally { + is = fileSystem.read(fsPath) + if (position > 0) is.skip(position) + breakable { + while (lengthLeft > 0) { + val readed = is.read(tub) + val useful = Math.min(readed, lengthLeft) + if (useful < 0) break() + lengthLeft -= useful + executionCode.append(new String(tub, 0, useful, CHARSET)) + } + } + } { + if (fileSystem != null) fileSystem.close() + IOUtils.closeQuietly(is) + } + queryTask.setExecutionCode(executionCode.toString()) + } + + private def getCodeStorePath(user: String): String = { + val date: String = DateFormatUtils.format(new Date, "yyyyMMdd") + s"${CODE_STORE_PREFIX.getValue}${user}${CODE_STORE_SUFFIX.getValue}/executionCode/${date}/_scripts" + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
