Commit 4ef683ae authored by sml's avatar sml

Merge branch 'dmp_dev' of

http://gitlab.ioubuy.cn/yaobenzhang/jz-dmp-service.git into dmp_dev

Conflicts:
	src/main/java/com/jz/dmp/modules/service/DmpDevelopTaskService.java
	src/main/java/com/jz/dmp/modules/service/impl/DmpDevelopTaskServiceImpl.java
parents b7f8dfec 5bf4ed26
......@@ -33,8 +33,33 @@ public class CommConstant {
/***************************************************/
//task_type任务类型
public static final String TASK_TYPE_OFFLINE = "2";
public static final String TASK_TYPE_DEVELOP = "1";//开发任务
public static final String TASK_TYPE_OFFLINE = "2";//离线任务
public static final String TASK_TYPE_DEVSHELL = "3";//SHELL任务
public static final String TASK_TYPE_DEVSQL = "4";//SQL任务
/***************************************************/
//tree type(节点类型)
public static final String TREE_TYPE_OFFLINE = "01";//离线任务
public static final String TREE_TYPE_REALTIMESYNC = "02";//实时同步
public static final String TREE_TYPE_DEVELOP = "03";//数据开发
public static final String TREE_TYPE_DEVSHELL = "04";//脚本开发
public static final String TREE_TYPE_DEVSQL = "05";//SQL开发
/***************************************************/
//子任务类型
public static final String WORK_TYPE_START = "start";//开始任务
public static final String WORK_TYPE_SHELL = "shell";//shell任务
public static final String WORK_TYPE_SQL = "sql";//sql任务
public static final String WORK_TYPE_SYNC = "sync";//离线任务
public static final String WORK_TYPE_SUBPROCESS = "subprocess";//子流程任务
public static final String WORK_TYPE_FTP = "ftp";//ftp下载任务
public static final String WORK_TYPE_UNZIPFILE = "unzipFile";//解压文件任务
public static final String WORK_TYPE_DOCTRANS = "docTrans";//文件转码任务
public static final String WORK_TYPE_HDFS = "hdfs";//hdfs上传任务
public static final String WORK_TYPE_STOP = "stop";//停止任务
/***************************************************/
//azkaban相关常量
public static final String AZKABAN_PROJECTNAME_PREFIX = "jz_workflow_new_"; //azkaban项目名称前缀
}
package com.jz.common.constant;
/**
* @ClassName: GatewayApiConstant
* @Description: 服务网关api地址
* @Author Bellamy
* @Date 2021/2/4
* @Version 1.0
*/
public class GatewayApiConstant {
//列表分页查询url
public static final String orgListPage = "/api/organization/listOrg";
//删除组织url
public static final String delOrg = "/api/organization/logoutOrg";
//新增组织url
public static final String addOrg = "/api/organization/add";
//编辑组织url
public static final String updateOrg = "/api/organization/update";
//根据组织id获取组织详情url
public static final String orgDetail = "/api/organization/getOrgDetail";
//授权给他人的API-列表分页查询url
public static final String apiAuthListPage = "/api/auth/auth-list";
//取消授权url
public static final String cancelApiAuth = "/api/auth/update-auth-info";
//发布的API-API列表分页查询url
public static final String apiListPage = "/api/interface/listApiInterface";
//删除APIurl
public static final String delApi = "/api/interface/delDMPApiInterface";
//根据apiid获取API详情url
public static final String getApiInfo = "/api/interface/getApiInterfaceDetail";
//授权给组织url
public static final String apiAuthToOrg = "/api/auth/dmp-auth-api";
//测试url
public static final String testApi = "/api/interface/apiTestApiInterface";
//查看日志url
public static final String checkApiLog = "/api/logging/listApiLog";
//授权模糊查询组织信息url
public static final String authOrgList = "/api/auth/getAuthNameList";
//保存API(第三方)基本信息url
public static final String addApiInfo = "/api/producer/addCustomApi";
//保存API(大数据查询/标签查询)url
public static final String addApiBigData = "/api/producer/makeBigDataApi";
//编辑API(第三方)基本信息url
public static final String updateApiInfo = "/api/producer/saveUpdate";
//API计量调用次数和执行时长url
public static final String countAPiCallStat = "/api/logging/countAPiCallStat";
//API计量--API未调用列表url
public static final String notCalledList = "/api/interface/listUnCallApiInterface";
//服务发布/服务取消url
public static final String apiPublishCancel = "/api/producer/optionApiIssueStatus";
}
......@@ -110,6 +110,13 @@ public class JsonResult<T> implements Serializable {
return result;
}
public static JsonResult<Object> error(ResultCode code) {
JsonResult<Object> result = new JsonResult<>();
result.setCode(code);
result.setMessage(code.msg());
return result;
}
public void setCodes(com.jz.dmp.agent.ResultCode code) {
this.code = code.val();
}
......
......@@ -28,4 +28,13 @@ public interface ExecutionFlowsMapper {
* @since 2021-01-14
*/
List<DataDevExamplesListDto> queryDevExamplesListPage(Map saveParams) throws Exception;
/**
* 获取执行实例的日志详情
*
* @return
* @author Bellamy
* @since 2021-02-03
*/
List<Map> queryExamplesLogByExecId(String execId) throws Exception;
}
package com.jz.dmp.modules.controller.DataIntegration;
import com.amazonaws.services.xray.model.Http;
import com.jz.common.constant.JsonResult;
import com.jz.common.constant.ResultCode;
import com.jz.common.page.BasePageBean;
......@@ -12,11 +13,13 @@ import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpRequest;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import javax.servlet.http.HttpServletRequest;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
......@@ -145,10 +148,11 @@ public class OfflineSynchController {
}
/**
* 根据taskId删除离线任务
* 根据taskId删除离线任务和任务树
*
* @return
* @author Bellamy
* @Date 2021/02/03
*/
@ApiOperation(value = "删除任务", notes = "删除任务")
@GetMapping(value = "/delTaskByTaskId")
......@@ -292,4 +296,5 @@ public class OfflineSynchController {
JsonResult list = offlineSynchService.addNewSynchTask(newSynchTaskReq);
return list;
}
}
......@@ -89,10 +89,12 @@ public class RealTimeSyncController {
DmpRealtimeSyncInfo dmpRealtimeSyncInfo = list.get(i);
String srcTopicName = dmpRealtimeSyncInfo.getSrcTopicName();
System.out.println(srcTopicName);
logger.info("############正常执行表数据id........" + ids[i]);
logger.info("############正常执行表数据id{}........" + ids[i]);
String shellPath = "/app/bigdata-app/scripts/trigger_straming.sh";
boolean flag = CmdUtils.callShell(shellPath, srcTopicName);
logger.info("############" + flag);
if(flag){
logger.info("############执行成功{}" + flag);
}
}
}
return new JsonResult();
......@@ -113,12 +115,7 @@ public class RealTimeSyncController {
if (StringUtils.isEmpty(realTaskId)) {
return new JsonResult(ResultCode.PARAMS_ERROR, "任务id不能为空!");
}
boolean jsonResult = dmpRealtimeSyncInfoService.deleteByrealTaskId(realTaskId);
if (jsonResult) {
return new JsonResult();
} else {
return new JsonResult(ResultCode.INTERNAL_SERVER_ERROR, "删除失败!");
}
return dmpRealtimeSyncInfoService.deleteByrealTaskId(realTaskId);
}
/**
......
......@@ -6,8 +6,10 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import com.jz.common.annotation.MethodCallLogPrint;
import com.jz.common.bean.BaseBeanResponse;
import com.jz.common.bean.BaseResponse;
import com.jz.common.bean.PageInfoResponse;
......@@ -18,6 +20,7 @@ import com.jz.dmp.modules.model.DmpDevelopTask;
import com.jz.dmp.modules.service.DmpDevelopTaskService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiOperation;
/**
......@@ -74,19 +77,145 @@ public class DmpDevelopTaskController {
return baseBeanResponse;
}
/**任务流程保存提交
/**treeId查询任务开发
* @param dmpDevelopTaskRequest
* @return
*/
@RequestMapping(method = RequestMethod.POST, value = "/flowSubmit")
@ApiOperation(value = "任务流程保存提交", notes = "任务流程保存提交")
public BaseResponse flowSubmit(@RequestBody DmpDevelopTask dmpDevelopTask, HttpServletRequest httpRequest){
@MethodCallLogPrint
@RequestMapping(method = RequestMethod.GET, value = "/findByTreeId")
@ApiOperation(value = "treeId查询任务开发", notes = "treeId查询任务开发")
@ApiImplicitParam(name = "treeId", value = "treeId")
public BaseBeanResponse<DmpDevelopTaskDto> view(@RequestParam(name = "treeId", required = true ) Integer treeId, HttpServletRequest httpRequest){
BaseBeanResponse<DmpDevelopTaskDto> baseBeanResponse = new BaseBeanResponse<DmpDevelopTaskDto>();
try {
baseBeanResponse = dmpDevelopTaskService.findByTreeId(treeId, httpRequest);
} catch (Exception e) {
baseBeanResponse.setMessage("请求失败");
baseBeanResponse.setCode(StatuConstant.FAILURE_CODE);
e.printStackTrace();
}
return baseBeanResponse;
}
/**修改任务开发
* @param dmpDevelopTaskRequest
* @return
*/
@MethodCallLogPrint
@RequestMapping(method = RequestMethod.POST, value = "/edit")
@ApiOperation(value = "修改任务开发", notes = "修改任务开发")
public BaseBeanResponse<DmpDevelopTask> edit(@RequestBody DmpDevelopTask dmpDevelopTask, HttpServletRequest httpRequest){
BaseBeanResponse<DmpDevelopTask> baseBeanResponse = new BaseBeanResponse<DmpDevelopTask>();
try {
baseBeanResponse = dmpDevelopTaskService.edit(dmpDevelopTask, httpRequest);
} catch (Exception e) {
baseBeanResponse.setMessage("修改失败");
baseBeanResponse.setCode(StatuConstant.FAILURE_CODE);
e.printStackTrace();
}
return baseBeanResponse;
}
/**任务流程发布接口
* @param dmpDevelopTaskRequest
* @return
*/
@RequestMapping(method = RequestMethod.GET, value = "/flowSubmit")
@ApiOperation(value = "任务流程发布接口", notes = "任务流程发布接口")
@ApiImplicitParam(name = "treeId", value = "发布任务树主键")
public BaseResponse flowSubmit(@RequestParam(value = "treeId", required = true) Long treeId, HttpServletRequest httpRequest){
BaseResponse baseResponse = new BaseResponse();
try {
baseResponse = dmpDevelopTaskService.flowSubmit(treeId, httpRequest);
} catch (Exception e) {
baseResponse.setMessage("任务流程发布失败");
baseResponse.setCode(StatuConstant.FAILURE_CODE);
e.printStackTrace();
}
return baseResponse;
}
/**SHELL/SQL/离线任务发布接口
* @param treeId
* @return
*/
@RequestMapping(method = RequestMethod.GET, value = "/taskPublish")
@ApiOperation(value = "SHELL/SQL/离线任务发布接口", notes = "SHELL/SQL/离线任务发布接口")
@ApiImplicitParam(name = "treeId", value = "发布任务树主键")
public BaseResponse taskPublish(@RequestParam(value = "treeId", required = true) Long treeId, HttpServletRequest httpRequest){
BaseResponse baseResponse = new BaseResponse();
try {
baseResponse = dmpDevelopTaskService.flowSubmit(dmpDevelopTask, httpRequest);
baseResponse = dmpDevelopTaskService.taskPublish(treeId, httpRequest);
} catch (Exception e) {
baseResponse.setMessage("SHELL/SQL/离线任务发布失败");
baseResponse.setCode(StatuConstant.FAILURE_CODE);
e.printStackTrace();
}
return baseResponse;
}
/**任务立即运行接口
* @param treeId
* @return
*/
@RequestMapping(method = RequestMethod.GET, value = "/taskAzkabanRun")
@ApiOperation(value = "任务立即运行接口", notes = "任务立即运行接口")
@ApiImplicitParam(name = "treeId", value = "任务树主键")
public BaseResponse taskAzkabanRun(@RequestParam(value = "treeId", required = true) Long treeId, HttpServletRequest httpRequest){
BaseResponse baseResponse = new BaseResponse();
try {
baseResponse = dmpDevelopTaskService.taskAzkabanRun(treeId, httpRequest);
} catch (Exception e) {
baseResponse.setMessage("任务运行失败");
baseResponse.setCode(StatuConstant.FAILURE_CODE);
e.printStackTrace();
}
return baseResponse;
}
/**任务立即运行停止
* @param treeId
* @return
*/
@RequestMapping(method = RequestMethod.GET, value = "/taskAzkabanStop")
@ApiOperation(value = "任务立即运行停止接口", notes = "任务立即运行停止接口")
@ApiImplicitParam(name = "treeId", value = "任务树主键")
public BaseBeanResponse<String> taskAzkabanStop(@RequestParam(value = "treeId", required = true) Long treeId, HttpServletRequest httpRequest){
BaseBeanResponse<String> baseBeanResponse = new BaseBeanResponse<String>();
try {
baseBeanResponse = dmpDevelopTaskService.taskAzkabanStop(treeId, httpRequest);
} catch (Exception e) {
baseBeanResponse.setMessage("任务运行失败");
baseBeanResponse.setCode(StatuConstant.FAILURE_CODE);
e.printStackTrace();
}
return baseBeanResponse;
}
/**软删除任务开发
* @param id
* @return
*/
@MethodCallLogPrint
@RequestMapping(method = RequestMethod.GET, value = "/softDeleteByTreeId")
@ApiOperation(value = "软删除任务开发", notes = "软删除任务开发")
@ApiImplicitParam(name = "treeId", value = "任务开发主键")
public BaseResponse softDeleteByTreeId(@RequestParam(name = "treeId", required = true ) Integer treeId, HttpServletRequest httpRequest){
BaseResponse baseResponse = new BaseResponse();
try {
baseResponse = dmpDevelopTaskService.softDeleteByTreeId(treeId, httpRequest);
} catch (Exception e) {
baseResponse.setMessage("任务流程保存提交失败");
baseResponse.setMessage("软删除失败");
baseResponse.setCode(StatuConstant.FAILURE_CODE);
e.printStackTrace();
}
......
......@@ -82,9 +82,14 @@ public class DmpNavigationTreeController {
@ApiOperation(value = "新增DMP资源导航树", notes = "新增DMP资源导航树")
public BaseBeanResponse<DmpNavigationTree> add(@RequestBody DmpNavigationTree dmpNavigationTree, HttpServletRequest httpRequest) throws Exception {
BaseBeanResponse<DmpNavigationTree> baseBeanResponse = new BaseBeanResponse<DmpNavigationTree>();
//校验参数
//
//树名称去重
DmpNavigationTreeRequest dmpNavigationTreeRequest = new DmpNavigationTreeRequest();
dmpNavigationTreeRequest.setName(dmpNavigationTree.getName());
dmpNavigationTreeRequest.setProjectId(dmpNavigationTree.getProjectId());
BaseBeanResponse<DmpNavigationTreeDto> baseBeanResponseRe = dmpNavigationTreeService.findList(dmpNavigationTreeRequest, httpRequest);
if (baseBeanResponseRe.getDatas() != null && baseBeanResponseRe.getDatas().size() > 0) {
baseBeanResponse.setCode(StatuConstant.CODE_ERROR_PARAMETER);
......
package com.jz.dmp.modules.controller;
import com.jz.dmp.modules.model.DmpTableColumn;
import com.jz.dmp.modules.service.DmpTableColumnService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
/**
* 项目表字段(DmpTableColumn)表控制层
*
* @author Bellamy
* @since 2020-12-23 16:34:56
*/
@RestController
@RequestMapping("/dmpTableColumn")
public class DmpTableColumnController {
/**
* 服务对象
*/
@Autowired
private DmpTableColumnService dmpTableColumnService;
}
\ No newline at end of file
package com.jz.dmp.modules.controller;
import com.jz.dmp.modules.model.DmpTable;
import com.jz.dmp.modules.service.DmpTableService;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
/**
* 项目表信息(DmpTable)表控制层
*
* @author Bellamy
* @since 2020-12-23 15:32:57
*/
@RestController
@RequestMapping("dmpTable")
public class DmpTableController {
/**
* 服务对象
*/
@Resource
private DmpTableService dmpTableService;
}
\ No newline at end of file
package com.jz.dmp.modules.controller;
import com.jz.dmp.modules.model.DmpTableFieldMapping;
import com.jz.dmp.modules.service.DmpTableFieldMappingService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
/**
* 项目表字段类型映射(DmpTableFieldMapping)表控制层
*
* @author Bellamy
* @since 2020-12-23 16:34:55
*/
@RestController
@RequestMapping("dmpTableFieldMapping")
public class DmpTableFieldMappingController{
/**
* 服务对象
*/
@Autowired
private DmpTableFieldMappingService dmpTableFieldMappingService;
}
\ No newline at end of file
package com.jz.dmp.modules.controller;
import com.jz.dmp.modules.model.DmpTableFieldSchema;
import com.jz.dmp.modules.service.DmpTableFieldSchemaService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
/**
* 项目表字段类型(DmpTableFieldSchema)表控制层
*
* @author Bellamy
* @since 2020-12-23 16:34:56
*/
@RestController
@RequestMapping("dmpTableFieldSchema")
public class DmpTableFieldSchemaController{
/**
* 服务对象
*/
@Autowired
private DmpTableFieldSchemaService dmpTableFieldSchemaService;
}
\ No newline at end of file
package com.jz.dmp.modules.controller;
import com.jz.dmp.modules.service.DvRuleTService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* ???(DvRuleT)表控制层
*
* @author Bellamy
* @since 2020-12-24 10:56:18
*/
@RestController
@RequestMapping("/dvRuleT")
public class DvRuleTController {
/**
* 服务对象
*/
@Autowired
private DvRuleTService dvRuleTService;
}
\ No newline at end of file
......@@ -56,6 +56,7 @@ public class MyDmpDevelopTaskHistoryConverter {
dmpDevelopTaskHistory.setTaskCreateTime(dmpDevelopTask.getCreateTime());
dmpDevelopTaskHistory.setTaskUpdateUserId(dmpDevelopTask.getUpdateUserId());
dmpDevelopTaskHistory.setTaskUpdateTime(dmpDevelopTask.getUpdateTime());
dmpDevelopTaskHistory.setId(null);
return dmpDevelopTaskHistory;
}
......
......@@ -58,5 +58,29 @@ public class DmpDevExamplesController {
return pageInfo;
}
/**
* 获取执行实例的日志详情
*
* @return
* @author Bellamy
* @since 2021-02-03
*/
@ApiOperation(value = "执行实例日志详情", notes = "执行实例日志详情")
@GetMapping(value = "/log")
@ApiImplicitParam(name = "execId", value = "执行实例ID", required = true)
public JsonResult getExamplesLog(@RequestParam String execId) {
if (StringUtils.isEmpty(execId)) {
return JsonResult.error(ResultCode.PARAMS_ERROR, "执行实例ID不能为空!");
}
JsonResult result = new JsonResult();
try {
result = dmpDevelopTaskService.queryExamplesLogByExecId(execId);
} catch (Exception e) {
result.setCode(ResultCode.INTERNAL_SERVER_ERROR);
result.setMessage("查询失败!");
e.printStackTrace();
}
return result;
}
}
......@@ -240,4 +240,29 @@ public class DmpApiMangeController {
}
return jsonResult;
}
/**
* 服务发布/服务取消
*
* @author Bellamy
* @since 2021-02-07
*/
@ApiOperation(value = "服务发布/服务取消", notes = "服务发布/服务取消")
@GetMapping(value = "/apiPublishCancel")
@ApiImplicitParam(name = "id", value = "id", required = true)
public JsonResult apiPublishCancel(@RequestParam String id, @RequestParam(name = "optStatus") Boolean optStatus) {
if (StringUtils.isEmpty(id)) {
return JsonResult.error(ResultCode.PARAMS_ERROR, "id不能为空!");
}
JsonResult jsonResult = new JsonResult();
try {
jsonResult = dmpApiMangeService.apiPublishCancel(id, optStatus);
} catch (Exception e) {
jsonResult.setMessage(e.getMessage());
jsonResult.setCode(ResultCode.INTERNAL_SERVER_ERROR);
e.printStackTrace();
}
return jsonResult;
}
}
\ No newline at end of file
......@@ -8,6 +8,13 @@ import io.swagger.annotations.ApiModelProperty;
* @author ybz
*
*/
/**
* @ClassName: DmpProjectConfigInfoDto
* @Description: TODO(这里用一句话描述这个类的作用)
* @author ybz
* @date 2021年2月4日
*
*/
@ApiModel(value = "项目配置表Dto", description = "项目配置表Dto")
public class DmpProjectConfigInfoDto extends DmpProjectConfigInfo {
......@@ -26,5 +33,4 @@ public class DmpProjectConfigInfoDto extends DmpProjectConfigInfo {
public void setDmpPublicConfigInfoDto(DmpPublicConfigInfoDto dmpPublicConfigInfoDto) {
this.dmpPublicConfigInfoDto = dmpPublicConfigInfoDto;
}
}
......@@ -72,4 +72,5 @@ public interface DmpDevelopTaskDao {
* @since 2021-02-01
*/
DmpDevelopTask selectTaskById(@Param("taskId") String taskId) throws Exception;
}
......@@ -4,6 +4,7 @@ import com.jz.dmp.modules.model.DmpNavigationTree;
import org.apache.ibatis.annotations.Param;
import java.util.List;
import java.util.Map;
/**
* DMP资源导航树(DmpNavigationTree)表数据库访问层
......@@ -112,4 +113,13 @@ public interface DmpNavigationTreeDao {
* @throws Exception
*/
public int insertSelective(DmpNavigationTree dmpNavigationTree) throws Exception;
/**
* 假删除任务树
*
* @param params 查询起始位置
* @return
* @Date 2021/02/03
*/
int deleteByTreeId(Map params) throws Exception;
}
\ No newline at end of file
......@@ -29,9 +29,9 @@ public class DmpDevelopTask implements Serializable {
@ApiModelProperty(value = "数据源ID")
private Integer datasourceId;
/**
* 任务类型(1,开发任务;2,离线任务)
* 任务类型(1,开发任务;2,离线任务;3,SHELL任务;4,SQL任务
*/
@ApiModelProperty(value = "任务类型(1,开发任务;2,离线任务)")
@ApiModelProperty(value = "任务类型(1,开发任务;2,离线任务;3,SHELL任务;4,SQL任务)")
private String taskType;
/**
* 类型
......@@ -75,9 +75,9 @@ public class DmpDevelopTask implements Serializable {
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
private Date createTime;
/**
* 创建用户ID
* 修改用户ID
*/
@ApiModelProperty(value = "创建用户ID")
@ApiModelProperty(value = "修改用户ID")
private String updateUserId;
/**
* 数据更新时间
......@@ -122,6 +122,12 @@ public class DmpDevelopTask implements Serializable {
@ApiModelProperty(value = "版本")
private String version;
/**
* 发布版本
*/
@ApiModelProperty(value = "发布版本")
private String publishVersion;
@ApiModelProperty(value = "是否压缩")
private Integer isGziped;
......@@ -343,6 +349,14 @@ public class DmpDevelopTask implements Serializable {
this.version = version;
}
public String getPublishVersion() {
return publishVersion;
}
public void setPublishVersion(String publishVersion) {
this.publishVersion = publishVersion;
}
public Integer getIsGziped() {
return isGziped;
}
......
......@@ -28,9 +28,9 @@ public class DmpNavigationTree implements Serializable {
@ApiModelProperty(value = "树类别(2:开发任务,3:脚本任务)")
private String category;
/**
* 树类型(01:离线同步,02:实时同步,03:数据开发)
* 树类型(01:离线同步,02:实时同步,03:数据开发,04:脚本开发,05:SQL开发
*/
@ApiModelProperty(value = "树类型(01:离线同步,02:实时同步,03:数据开发)")
@ApiModelProperty(value = "树类型(01:离线同步,02:实时同步,03:数据开发,04:脚本开发,05:SQL开发)")
private String type;
/**
* 名称
......
......@@ -157,6 +157,12 @@ public class DmpRealtimeSyncInfo implements Serializable {
@ApiModelProperty(value = "版本号")
private String version;
/**
* treeId
*/
@ApiModelProperty(value = "treeId")
private String treeId;
public Integer getId() {
return id;
}
......@@ -396,4 +402,12 @@ public class DmpRealtimeSyncInfo implements Serializable {
public void setVersion(String version) {
this.version = version;
}
public String getTreeId() {
return treeId;
}
public void setTreeId(String treeId) {
this.treeId = treeId;
}
}
\ No newline at end of file
......@@ -83,4 +83,11 @@ public interface DmpApiMangeService {
* @since 2021-01-19
*/
JsonResult getAuthOrgList(String key) throws Exception;
/**
* @Description:服务发布/服务取消
* @author Bellamy
* @since 2021-02-07
*/
JsonResult apiPublishCancel(String id, Boolean optStatus) throws Exception;
}
......@@ -95,6 +95,7 @@ public interface DmpDevelopTaskService {
*/
public BaseBeanResponse<DmpDevelopTask> edit(DmpDevelopTask dmpDevelopTask, HttpServletRequest httpRequest)throws Exception;
/**
* @Title: flowSubmit
* @Description: TODO(任务流程发布到azkaban)
......@@ -107,17 +108,6 @@ public interface DmpDevelopTaskService {
*/
public BaseResponse flowSubmit(Long treeId, HttpServletRequest httpRequest)throws Exception;
/**
* @Title: getExecXmlFileName
* @Description: TODO(根据任务treeId获取xmlFileName)
* @param @param syncTaskTreeId
* @param @return
* @param @throws Exception 参数
* @return String 返回类型
* @throws
*/
public String getExecXmlFileName(Long syncTaskTreeId)throws Exception;
/**
* @Title: taskAzkabanRun
* @Description: TODO(运行任务)
......@@ -165,6 +155,16 @@ public interface DmpDevelopTaskService {
*/
public BaseResponse softDeleteByTreeId(Integer treeId, HttpServletRequest httpRequest)throws Exception;
/**
* @param @param syncTaskTreeId
* @param @return
* @param @throws Exception 参数
* @return String 返回类型
* @throws
* @Title: getExecXmlFileName
* @Description: TODO(根据任务treeId获取xmlFileName)
*/
public String getExecXmlFileName(Long syncTaskTreeId) throws Exception;
/**
* 获取执行实例的日志详情
......
......@@ -134,7 +134,7 @@ public interface DmpRealtimeSyncInfoService {
* @return
* @author Bellamy
*/
boolean deleteByrealTaskId(String realTaskId) throws Exception;
JsonResult deleteByrealTaskId(String realTaskId) throws Exception;
/**
* 批量上下线
......
......@@ -51,10 +51,11 @@ public interface OfflineSynchService {
JsonResult taskRunNowByTaskId(String taskId) throws Exception;
/**
* 根据taskId删除离线任务
* 根据taskId删除离线任务和任务树
*
* @return
* @author Bellamy
* @Date 2021/02/03
*/
JsonResult delTaskByTaskId(String taskId) throws Exception;
......
package com.jz.dmp.modules.service.impl;
import com.alibaba.fastjson.JSONObject;
import com.jz.common.constant.GatewayApiConstant;
import com.jz.common.constant.JsonResult;
import com.jz.common.constant.ResultCode;
import com.jz.common.utils.web.HttpClientUtils;
import com.jz.dmp.modules.controller.dataService.bean.*;
import com.jz.dmp.modules.service.DmpApiMangeService;
import com.jz.dmp.modules.service.DmpOrgMangeService;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.NameValuePair;
import org.apache.http.message.BasicNameValuePair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
......@@ -30,33 +28,6 @@ public class DmpApiMangeServiceImpl implements DmpApiMangeService {
private static Logger logger = LoggerFactory.getLogger(DmpApiMangeServiceImpl.class);
//授权给他人的API-列表分页查询url
private static final String apiAuthListPage = "/api/auth/auth-list";
//取消授权url
private static final String cancelApiAuth = "/api/auth/update-auth-info";
//发布的API-API列表分页查询url
private static final String apiListPage = "/api/interface/listApiInterface";
//删除APIurl
private static final String delApi = "/api/interface/delDMPApiInterface";
//根据apiid获取API详情url
private static final String getApiInfo = "/api/interface/getApiInterfaceDetail";
//授权给组织url
private static final String apiAuthToOrg = "/api/auth/dmp-auth-api";
//测试url
private static final String testApi = "/api/interface/apiTestApiInterface";
//查看日志url
private static final String checkApiLog = "/api/logging/listApiLog";
//授权模糊查询组织信息url
private static final String authOrgList = "/api/auth/getAuthNameList";
@Value("${spring.gateway-url}")
private String gatewayUrl;
......@@ -69,7 +40,7 @@ public class DmpApiMangeServiceImpl implements DmpApiMangeService {
@Override
public JsonResult queryApiAuthListPage(AuthListInfoReq req) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + apiAuthListPage;
String url = gatewayUrl + GatewayApiConstant.apiAuthListPage;
String resultData = HttpClientUtils.post(url, JSONObject.toJSONString(req));
if (StringUtils.isEmpty(resultData)) {
throw new RuntimeException("查询失败!");
......@@ -99,7 +70,7 @@ public class DmpApiMangeServiceImpl implements DmpApiMangeService {
@Override
public JsonResult cancelApiAuth(String id) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + cancelApiAuth;
String url = gatewayUrl + GatewayApiConstant.cancelApiAuth;
Map params = new HashMap();
params.put("id", id);
String resultData = HttpClientUtils.post(url, JSONObject.toJSONString(params));
......@@ -131,7 +102,7 @@ public class DmpApiMangeServiceImpl implements DmpApiMangeService {
@Override
public JsonResult queryApiListPage(ApiInterfaceInfoListReq req) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + apiListPage;
String url = gatewayUrl + GatewayApiConstant.apiListPage;
String resultData = HttpClientUtils.post(url, JSONObject.toJSONString(req));
if (StringUtils.isEmpty(resultData)) {
throw new RuntimeException("查询失败!");
......@@ -161,7 +132,7 @@ public class DmpApiMangeServiceImpl implements DmpApiMangeService {
@Override
public JsonResult delApiInfo(String apiKey, String type) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + delApi;
String url = gatewayUrl + GatewayApiConstant.delApi;
Map params = new HashMap();
params.put("apiKey", apiKey);
params.put("type", type);
......@@ -193,7 +164,7 @@ public class DmpApiMangeServiceImpl implements DmpApiMangeService {
@Override
public JsonResult queryApiInfoByApiId(String id) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + getApiInfo;
String url = gatewayUrl + GatewayApiConstant.getApiInfo;
Map params = new HashMap();
params.put("id", id);
......@@ -226,7 +197,7 @@ public class DmpApiMangeServiceImpl implements DmpApiMangeService {
@Override
public JsonResult addApiAuthToOrg(AuthUserApiReq req) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + apiAuthToOrg;
String url = gatewayUrl + GatewayApiConstant.apiAuthToOrg;
String resultData = HttpClientUtils.post(url, JSONObject.toJSONString(req));
if (StringUtils.isEmpty(resultData)) {
throw new RuntimeException("保存失败!");
......@@ -257,7 +228,7 @@ public class DmpApiMangeServiceImpl implements DmpApiMangeService {
@Override
public JsonResult apiTestInfo(String apiKey) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + testApi;
String url = gatewayUrl + GatewayApiConstant.testApi;
Map params = new HashMap();
params.put("apiKey", apiKey);
......@@ -289,7 +260,7 @@ public class DmpApiMangeServiceImpl implements DmpApiMangeService {
@Override
public JsonResult checkApiLogInfo(LogInfoListReq req) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + checkApiLog;
String url = gatewayUrl + GatewayApiConstant.checkApiLog;
String resultData = HttpClientUtils.post(url, JSONObject.toJSONString(req));
if (StringUtils.isEmpty(resultData)) {
throw new RuntimeException("查询失败!");
......@@ -320,7 +291,7 @@ public class DmpApiMangeServiceImpl implements DmpApiMangeService {
@Override
public JsonResult getAuthOrgList(String key) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + authOrgList;
String url = gatewayUrl + GatewayApiConstant.authOrgList;
Map params = new HashMap();
params.put("key", key);
......@@ -343,4 +314,38 @@ public class DmpApiMangeServiceImpl implements DmpApiMangeService {
return result;
}
/**
* @param id
* @param optStatus
* @Description:服务发布/服务取消
* @author Bellamy
* @since 2021-02-07
*/
@Override
public JsonResult apiPublishCancel(String id, Boolean optStatus) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + GatewayApiConstant.apiPublishCancel;
Map params = new HashMap();
params.put("id", id);
params.put("optStatus", optStatus);
String returnData = HttpClientUtils.getJsonForParam(url, params);
if (StringUtils.isEmpty(returnData)) {
throw new RuntimeException("失败!");
}
logger.info("#################响应结果{}" + returnData);
Map map = JSONObject.parseObject(returnData);
if (map.containsKey("code")) {
if ("200".equals(map.get("code").toString())) {
return JsonResult.ok();
}
}
if (map.containsKey("message")) {
logger.info(map.get("message").toString());
result.setMessage(map.get("message").toString());
result.setCode(ResultCode.INTERNAL_SERVER_ERROR);
}
return result;
}
}
......@@ -9,12 +9,10 @@ import com.jz.common.utils.web.SessionUtils;
import com.jz.dmp.modules.controller.DataIntegration.bean.SourceDbNameListDto;
import com.jz.dmp.modules.controller.dataService.bean.ApiInterfaceInfoListReq;
import com.jz.dmp.modules.controller.dataService.bean.ApiInterfaceReq;
import com.jz.common.constant.GatewayApiConstant;
import com.jz.dmp.modules.controller.dataService.bean.MakeBigDataApiReq;
import com.jz.dmp.modules.dao.DvRuleTDao;
import com.jz.dmp.modules.dao.OfflineSynchDao;
import com.jz.dmp.modules.model.DvRuleT;
import com.jz.dmp.modules.service.DmpApiServiceMangeService;
import com.jz.dmp.modules.service.DvRuleTService;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -37,21 +35,6 @@ public class DmpApiServiceMangeServiceImpl implements DmpApiServiceMangeService
private static Logger logger = LoggerFactory.getLogger(DmpApiServiceMangeServiceImpl.class);
//保存API(第三方)基本信息url
private static final String addApiInfo = "/api/producer/addCustomApi";
//保存API(大数据查询/标签查询)url
private static final String addApiBigData = "/api/producer/makeBigDataApi";
//编辑API(第三方)基本信息url
private static final String updateApiInfo = "/api/producer/saveUpdate";
//API计量调用次数和执行时长url
private static final String countAPiCallStat = "/api/logging/countAPiCallStat";
//API计量--API未调用列表url
private static final String notCalledList = "/api/interface/listUnCallApiInterface";
@Value("${spring.gateway-url}")
private String gatewayUrl;
......@@ -70,7 +53,7 @@ public class DmpApiServiceMangeServiceImpl implements DmpApiServiceMangeService
@Override
public JsonResult saveApiInfo(ApiInterfaceReq req) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + addApiInfo;
String url = gatewayUrl + GatewayApiConstant.addApiInfo;
req.setCreateUser(SessionUtils.getCurrentUserName());
String resultData = HttpClientUtils.post(url, JSONObject.toJSONString(req));
......@@ -114,7 +97,7 @@ public class DmpApiServiceMangeServiceImpl implements DmpApiServiceMangeService
req.setApiKey(apiKey);
req.setCreateUser(SessionUtils.getCurrentUserName());
String url = gatewayUrl + addApiBigData;//保存API(大数据查询/标签查询)
String url = gatewayUrl + GatewayApiConstant.addApiBigData;//保存API(大数据查询/标签查询)
String resultData = HttpClientUtils.post(url, JSONObject.toJSONString(req));
if (StringUtils.isEmpty(resultData)) {
throw new RuntimeException("保存失败!");
......@@ -145,7 +128,7 @@ public class DmpApiServiceMangeServiceImpl implements DmpApiServiceMangeService
@Override
public JsonResult updateApiInfo(ApiInterfaceReq req) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + updateApiInfo;
String url = gatewayUrl + GatewayApiConstant.updateApiInfo;
req.setCreateUser(SessionUtils.getCurrentUserName());
String resultData = HttpClientUtils.post(url, JSONObject.toJSONString(req));
......@@ -187,7 +170,7 @@ public class DmpApiServiceMangeServiceImpl implements DmpApiServiceMangeService
//编辑API(大数据查询/标签查询)
JsonResult result = new JsonResult();
req.setCreateUser(SessionUtils.getCurrentUserName());
String url = gatewayUrl + addApiBigData;
String url = gatewayUrl + GatewayApiConstant.addApiBigData;
String resultData = HttpClientUtils.post(url, JSONObject.toJSONString(req));
if (StringUtils.isEmpty(resultData)) {
throw new RuntimeException("编辑失败!");
......@@ -217,7 +200,7 @@ public class DmpApiServiceMangeServiceImpl implements DmpApiServiceMangeService
@Override
public JsonResult getCountAPiCallStat() throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + countAPiCallStat;
String url = gatewayUrl + GatewayApiConstant.countAPiCallStat;
Map params = new HashMap();
params.put("date", DateUtils.currentDate());
......@@ -247,7 +230,7 @@ public class DmpApiServiceMangeServiceImpl implements DmpApiServiceMangeService
* @author Bellamy
*/
@Override
public JsonResult querygSourceDbList(Integer projectId,String dbName) throws Exception {
public JsonResult querygSourceDbList(Integer projectId, String dbName) throws Exception {
Map map = new HashMap();
map.put("projectId", projectId); //项目id
map.put("dbName", dbName);
......@@ -278,7 +261,7 @@ public class DmpApiServiceMangeServiceImpl implements DmpApiServiceMangeService
@Override
public JsonResult queryApiNotCalledListPage(ApiInterfaceInfoListReq req) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + notCalledList;
String url = gatewayUrl + GatewayApiConstant.notCalledList;
String resultData = HttpClientUtils.post(url, JSONObject.toJSONString(req));
if (StringUtils.isEmpty(resultData)) {
throw new RuntimeException("查询失败!");
......
package com.jz.dmp.modules.service.impl;
import com.alibaba.fastjson.JSONObject;
import com.jcraft.jsch.Session;
import com.jz.common.constant.GatewayApiConstant;
import com.jz.common.constant.JsonResult;
import com.jz.common.constant.ResultCode;
import com.jz.common.utils.web.HttpClientUtils;
import com.jz.common.utils.web.SessionUtils;
import com.jz.dmp.modules.controller.dataService.bean.OrganizationManageAddReq;
import com.jz.dmp.modules.controller.dataService.bean.OrganizationManageDetailQueryReq;
import com.jz.dmp.modules.controller.dataService.bean.OrganizationManageListQueryReq;
import com.jz.dmp.modules.controller.dataService.bean.OrganizationManageUpdateReq;
import com.jz.dmp.modules.controller.dataService.bean.*;
import com.jz.dmp.modules.service.DmpOrgMangeService;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
......@@ -32,21 +29,6 @@ public class DmpOrgMangeServiceImpl implements DmpOrgMangeService {
private static Logger logger = LoggerFactory.getLogger(DmpOrgMangeServiceImpl.class);
//列表分页查询url
private static final String orgListPage = "/api/organization/listOrg";
//删除组织url
private static final String delOrg = "/api/organization/logoutOrg";
//新增组织url
private static final String addOrg = "/api/organization/add";
//编辑组织url
private static final String updateOrg = "/api/organization/update";
//根据组织id获取组织详情url
private static final String orgDetail = "/api/organization/getOrgDetail";
@Value("${spring.gateway-url}")
private String gatewayUrl;
......@@ -60,7 +42,7 @@ public class DmpOrgMangeServiceImpl implements DmpOrgMangeService {
@Override
public JsonResult queryOrgListPage(OrganizationManageListQueryReq req) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + orgListPage;
String url = gatewayUrl + GatewayApiConstant.orgListPage;
if (StringUtils.isNotEmpty(req.getOrgName())) {
req.setOrgName(req.getOrgName().trim());
......@@ -96,7 +78,7 @@ public class DmpOrgMangeServiceImpl implements DmpOrgMangeService {
public JsonResult delOrgById(long id) throws Exception {
Map params = new HashMap();
params.put("id", id);
String url = gatewayUrl + delOrg;
String url = gatewayUrl + GatewayApiConstant.delOrg;
String returnData = HttpClientUtils.getJsonForParam(url, params);
if (StringUtils.isEmpty(returnData)) {
throw new RuntimeException("删除失败!");
......@@ -121,7 +103,7 @@ public class DmpOrgMangeServiceImpl implements DmpOrgMangeService {
@Override
public JsonResult addOrg(OrganizationManageAddReq req) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + addOrg;
String url = gatewayUrl + GatewayApiConstant.addOrg;
req.setCreateUser(SessionUtils.getCurrentUserName());
String returnData = HttpClientUtils.post(url, JSONObject.toJSONString(req));
if (StringUtils.isEmpty(returnData)) {
......@@ -153,7 +135,7 @@ public class DmpOrgMangeServiceImpl implements DmpOrgMangeService {
@Override
public JsonResult updateOrg(OrganizationManageUpdateReq req) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + addOrg;
String url = gatewayUrl + GatewayApiConstant.addOrg;
req.setCreateUser(SessionUtils.getCurrentUserName());
String returnData = HttpClientUtils.post(url, JSONObject.toJSONString(req));
......@@ -185,7 +167,7 @@ public class DmpOrgMangeServiceImpl implements DmpOrgMangeService {
@Override
public JsonResult getOrgInfoByOrgId(String id) throws Exception {
JsonResult result = new JsonResult();
String url = gatewayUrl + orgDetail;
String url = gatewayUrl + GatewayApiConstant.orgDetail;
OrganizationManageDetailQueryReq req = new OrganizationManageDetailQueryReq();
req.setId(Long.valueOf(id));
String returnData = HttpClientUtils.post(url, JSONObject.toJSONString(req));
......
package com.jz.dmp.modules.service.impl;
import com.amazonaws.services.dynamodbv2.xspec.M;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.jz.common.constant.JsonResult;
......@@ -13,6 +12,7 @@ import com.jz.common.utils.realTime.DBUtil;
import com.jz.common.utils.realTime.RestClient;
import com.jz.common.utils.web.SessionUtils;
import com.jz.dmp.modules.controller.DataIntegration.bean.*;
import com.jz.dmp.modules.dao.DmpNavigationTreeDao;
import com.jz.dmp.modules.dao.DmpProjectDao;
import com.jz.dmp.modules.dao.DmpRealtimeSyncHandleCountDao;
import com.jz.dmp.modules.dao.DmpRealtimeSyncInfoDao;
......@@ -67,6 +67,9 @@ public class DmpRealtimeSyncInfoServiceImpl implements DmpRealtimeSyncInfoServic
@Autowired
private DmpRealtimeSyncHandleCountDao dmpRealtimeSyncHandleCountDao;
@Autowired
private DmpNavigationTreeDao dmpNavigationTreeDao;
/**
* 通过ID查询单条数据
*
......@@ -308,18 +311,19 @@ public class DmpRealtimeSyncInfoServiceImpl implements DmpRealtimeSyncInfoServic
params.put("connectorSecurityFlag", dmpProjectSystemInfo.getKerberosIsenable()); //安全验证开关,是否启用KERBEROS
//connect1@http://172.18.104.130:9993/connectors
if (StringUtils.isNotEmpty(connectorUrl)) {
if (StringUtils.isEmpty(connectorUrl))
return JsonResult.error(ResultCode.PARAMS_ERROR, "connectorUrl不能为空!");
if (connectorUrl.contains("@")) {
connectorUrl = connectorUrl.split("@")[1];
}
}
//提交源到源的connector
Long realtimeId = submitDatasource2DatasourceToConnector(projectId, sourceDbInfo, targetDbInfo, dmpProjectSystemInfo, connectorUrl, params);
if (realtimeId != null) {
if (realtimeId == null) {
throw new RuntimeException("保存失败!");
}
//处理已选择的表信息
submitNoSelectTable(realtimeId, projectId, sourceDbInfo, targetDbInfo, dmpProjectSystemInfo, connectorUrl, params);
}
return new JsonResult();
return JsonResult.ok();
}
private Long submitDatasource2DatasourceToConnector(Long projectId, RealTimeSyncDataSourceModel sourceDbInfo, RealTimeSyncDataSourceModel targetDbInfo, DmpProjectSystemInfo dmpProjectSystemInfo, String connectorUrl, Map<String, Object> params) throws Exception {
......@@ -402,7 +406,7 @@ public class DmpRealtimeSyncInfoServiceImpl implements DmpRealtimeSyncInfoServic
logger.info("###################保存实时同步任务--结束 ################");
DmpRealtimeTaskHistory taskHistory = new DmpRealtimeTaskHistory();
BeanUtils.copyProperties(saveBody,taskHistory);
BeanUtils.copyProperties(saveBody, taskHistory);
taskHistory.setRealtimeSyncId(saveBody.getId());
dmpRealtimeSyncInfoDao.insertRealtimeHistory(taskHistory);
......@@ -604,7 +608,7 @@ public class DmpRealtimeSyncInfoServiceImpl implements DmpRealtimeSyncInfoServic
}
}
logger.info("###################处理已选择的表信息--结束################");
return new JsonResult();
return JsonResult.ok();
}
/**
......@@ -641,17 +645,18 @@ public class DmpRealtimeSyncInfoServiceImpl implements DmpRealtimeSyncInfoServic
params.put("connectorSecurityFlag", dmpProjectSystemInfo.getKerberosIsenable()); //安全验证开关,是否启用KERBEROS
//connect1@http://172.18.104.130:9993/connectors
if (StringUtils.isNotEmpty(connectorUrl)) {
if (StringUtils.isEmpty(connectorUrl)) {
return JsonResult.error(ResultCode.PARAMS_ERROR, "connectorUrl不能为空!");
}
if (connectorUrl.contains("@")) {
connectorUrl = connectorUrl.split("@")[1];
}
}
//提交源到源的connector
JsonResult realtimeId = updateDatasource2DatasourceToConnector(projectId, sourceDbInfo, targetDbInfo, dmpProjectSystemInfo, connectorUrl, params);
//编辑 已选择表信息
updateNoSelectTable(params);
return new JsonResult();
this.updateNoSelectTable(params);
return JsonResult.ok();
}
private JsonResult updateDatasource2DatasourceToConnector(Long projectId, RealTimeSyncDataSourceModel sourceDbInfo, RealTimeSyncDataSourceModel targetDbInfo, DmpProjectSystemInfo dmpProjectSystemInfo, String connectorUrl, Map<String, Object> params) throws Exception {
......@@ -719,6 +724,13 @@ public class DmpRealtimeSyncInfoServiceImpl implements DmpRealtimeSyncInfoServic
dmpRealtimeSyncInfoDao.update(saveBody);
logger.info("###################修改实时同步任务--结束 ################");
DmpRealtimeTaskHistory taskHistory = new DmpRealtimeTaskHistory();
BeanUtils.copyProperties(saveBody, taskHistory);
taskHistory.setRealtimeSyncId(saveBody.getId());
taskHistory.setCrePerson(SessionUtils.getCurrentUserId());
taskHistory.setCreateTime(new Date());
dmpRealtimeSyncInfoDao.insertRealtimeHistory(taskHistory);
Map blacklist = new HashMap();
blacklist.put("uptTime", new Date());
blacklist.put("uptPerson", SessionUtils.getCurrentUserId());
......@@ -800,19 +812,30 @@ public class DmpRealtimeSyncInfoServiceImpl implements DmpRealtimeSyncInfoServic
}
/**
* 批量删除数据源
* 删除实时同步任务
*
* @return
* @author Bellamy
* @since 2021-01-05
*/
@Override
@Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)
public boolean deleteByrealTaskId(String realTaskId) throws Exception {
public JsonResult deleteByrealTaskId(String realTaskId) throws Exception {
DmpRealtimeSyncInfo realtimeTask = dmpRealtimeSyncInfoDao.queryById(Integer.valueOf(realTaskId));
if (realtimeTask == null) {
return JsonResult.error(ResultCode.OPERATION_DATA_NO_EXIST);
}
Map map = new HashMap();
String[] ids = realTaskId.split(",");
map.put("ids", ids);
map.put("dataStatus", DelFlagEnum.YES.getValue());
return dmpRealtimeSyncInfoDao.deleteByrealTaskId(map) > 0;
int len = dmpRealtimeSyncInfoDao.deleteByrealTaskId(map);
if (len > 0) {
map.put("ids", realtimeTask.getTreeId());
dmpNavigationTreeDao.deleteByTreeId(map);
}
return JsonResult.ok();
}
/**
......
......@@ -19,6 +19,7 @@ import com.jz.common.utils.web.XmlUtils;
import com.jz.dmp.agent.DmpAgentResult;
import com.jz.dmp.modules.controller.DataIntegration.bean.*;
import com.jz.dmp.modules.controller.DataIntegration.bean.flow.FlowExecution;
import com.jz.dmp.modules.controller.bean.DmpNavigationTreeDto;
import com.jz.dmp.modules.dao.*;
import com.jz.dmp.modules.model.*;
import com.jz.dmp.modules.service.DmpDevelopTaskService;
......@@ -35,6 +36,7 @@ import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import springfox.documentation.spring.web.json.Json;
import javax.servlet.http.HttpServletRequest;
import java.io.File;
......@@ -246,29 +248,29 @@ public class OfflineSynchServiceImpl implements OfflineSynchService {
}
/**
* 批量删除离线任务
* 批量删除离线任务和任务树
*
* @author Bellamy
* @Date 2021/02/03
*/
@Override
@Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)
public JsonResult delTaskByTaskId(String taskId) throws Exception {
//通过taskId,查询任务和资源是否存在
Map map = dmpDevelopTaskDao.getDmpTaskAndTreeInfo(taskId);
if (map == null && map.size() == 0) {
return JsonResult.error(ResultCode.OPERATION_DATA_NO_EXIST);
}
Map params = new HashMap();
String[] ids = taskId.split(",");
params.put("ids", ids);
params.put("dataStatus", DelFlagEnum.YES.getValue());
dmpDevelopTaskDao.deleteTaskByTaskId(params);
//通过taskId,查询任务和资源是否存在
/*Map map = dmpDevelopTaskDao.getDmpTaskAndTreeInfo(taskId);
if (map == null) {
return new JsonResult(ResultCode.OPERATION_DATA_NO_EXIST);
}
if (StringUtils.isEmpty(map.get("treeId").toString())) {
return new JsonResult(ResultCode.OPERATION_DATA_NO_EXIST);
}*/
//dmpDevelopTaskDao.deleteNavigationTreeByTreeId(map.get("treeId").toString());
return new JsonResult(ResultCode.SUCCESS);
params.put("ids", map.get("treeId").toString());
dmpNavigationTreeDao.deleteByTreeId(params);
return JsonResult.ok();
}
/**
......@@ -673,7 +675,8 @@ public class OfflineSynchServiceImpl implements OfflineSynchService {
}
}
//保存时提交XML
return dmpDevelopTaskService.submitSyncing(task);
//dmpDevelopTaskService.submitSyncing(task);
return JsonResult.ok();
}
/**
......
......@@ -192,4 +192,22 @@
<if test="endTime != null and endTime != ''">#{endTime} >= and from_unixtime(submit_time/1000,'%Y-%m-%d %H:%i:%s')</if>
</select>
<!-- <resultMap type="java.util.HashMap" id="taskExamplesLog">
<result property="execId" column="exec_id" jdbcType="INTEGER"/>
<result property="name" column="name" jdbcType="VARCHAR"/>
<result property="log" column="log" jdbcType="BLOB" typeHandler="com.jz.common.persistence.CBTHandler"/>
<result property="uploadTime" column="upload_time" jdbcType="INTEGER"/>
</resultMap>-->
<!--获取执行实例的日志详情-->
<select id="queryExamplesLogByExecId" parameterType="string" resultType="java.util.Map">
select
exec_id as execId,
name,
log,
from_unixtime(upload_time/1000, '%Y-%m-%d %H:%i:%s') as uploadTime
from execution_logs
where 1=1 and exec_id=#{execId}
</select>
</mapper>
\ No newline at end of file
......@@ -706,10 +706,8 @@
<!-- 主键查询任务历史版本 -->
<select id="getMaxVersionByTaskId" resultType="java.lang.String">
select t.version from (
select MAX(CONVERT(REPLACE(RIGHT(version,LENGTH(version)-1),'.',''), DECIMAL) ), version
from dmp_develop_task_history where task_id = 0
) t
select MAX(CONVERT(version, DECIMAL)) AS version
from dmp_develop_task_history where task_id = #{taskId, jdbcType=INTEGER}
</select>
</mapper>
......@@ -18,17 +18,20 @@
<result column="flow_header" property="flowHeader"
jdbcType="VARCHAR" />
<result column="version" property="version" jdbcType="VARCHAR" />
<result column="publish_version" property="publishVersion" jdbcType="VARCHAR" />
<result column="data_status" property="dataStatus"
jdbcType="CHAR" />
<result column="tree_id" property="treeId" jdbcType="INTEGER" />
<result column="is_gziped" property="isGziped" jdbcType="INTEGER" />
<result column="name" property="name" jdbcType="VARCHAR" />
<result column="project_id" property="projectId" jdbcType="VARCHAR" />
<!-- <result column="chk_result" property="chkResult" javaType="VARCHAR"/> -->
<!-- <result column="sync_result" property="syncResult" javaType="VARCHAR"/> -->
</resultMap>
<sql id="FIND_ALL_COLUMN">
id,task_type,type,schedule_type,is_submit,task_desc,script,data_status,create_user_id,create_time,update_user_id,update_time,tree_id,flow_header,
flow_json, version,is_gziped
flow_json, version, publish_version, is_gziped
</sql>
<select id="getDmpTaskAndTreeInfo" parameterType="string" resultType="map">
......@@ -71,9 +74,9 @@
<!--新增所有列-->
<insert id="insert" keyProperty="id" useGeneratedKeys="true">
insert into dmp_develop_task(datasource_id, TASK_TYPE, TYPE, SCHEDULE_TYPE, IS_SUBMIT, TASK_DESC, SCRIPT, DATA_STATUS, CREATE_USER_ID, CREATE_TIME,
UPDATE_USER_ID, UPDATE_TIME, TREE_ID, CHK_RESULT, SYNC_RESULT, CHK_TIME, SYNC_TIME, FLOW_HEADER, FLOW_JSON, VERSION, IS_GZIPED,SOURCE_DB_NAME,SOURCE_TABLE_NAME,TARGET_DB_NAME,TARGET_TABLE_NAME)
UPDATE_USER_ID, UPDATE_TIME, TREE_ID, CHK_RESULT, SYNC_RESULT, CHK_TIME, SYNC_TIME, FLOW_HEADER, FLOW_JSON, VERSION, publish_version, IS_GZIPED,SOURCE_DB_NAME,SOURCE_TABLE_NAME,TARGET_DB_NAME,TARGET_TABLE_NAME)
values (#{datasourceId}, #{taskType}, #{type}, #{scheduleType}, #{isSubmit}, #{taskDesc}, #{data}, #{dataStatus}, #{createUserId}, #{createTime}, #{updateUserId},
#{updateTime}, #{treeId}, #{chkResult}, #{syncResult}, #{chkTime}, #{syncTime}, #{flowHeader}, #{flowJson}, #{version}, #{isGziped}, #{sourceDbName}, #{sourceTableName}, #{targetDbName}, #{targetTableName})
#{updateTime}, #{treeId}, #{chkResult}, #{syncResult}, #{chkTime}, #{syncTime}, #{flowHeader}, #{flowJson}, #{version}, #{publishVersion,jdbcType=VARCHAR}, #{isGziped}, #{sourceDbName}, #{sourceTableName}, #{targetDbName}, #{targetTableName})
</insert>
<update id="update">
......@@ -98,7 +101,7 @@
TASK_DESC = #{taskDesc},
</if>
<if test="script != null">
SCRIPT = #{data},
SCRIPT = #{script, jdbcType=BLOB},
</if>
<if test="dataStatus != null and dataStatus != ''">
DATA_STATUS = #{dataStatus},
......@@ -139,6 +142,9 @@
<if test="version != null and version != ''">
VERSION = #{version},
</if>
<if test="publishVersion != null and publishVersion != ''">
publish_version = #{publishVersion},,
</if>
<if test="isGziped != null">
IS_GZIPED = #{isGziped},
</if>
......@@ -147,11 +153,12 @@
</update>
<select id="selectTaskInfoByParam" parameterType="map" resultType="com.jz.dmp.modules.model.DmpDevelopTask">
select
ID, datasource_id, TASK_TYPE, TYPE, SCHEDULE_TYPE, IS_SUBMIT, TASK_DESC, SCRIPT, DATA_STATUS
, CREATE_USER_ID, CREATE_TIME, UPDATE_USER_ID, UPDATE_TIME, TREE_ID, CHK_RESULT, SYNC_RESULT, CHK_TIME, SYNC_TIME, FLOW_HEADER, FLOW_JSON, VERSION, IS_GZIPED
from dmp_develop_task
where data_status ='1' and TREE_ID = #{treeId}
SELECT
task.id,task.task_type,task.type,task.schedule_type,task.is_submit,task.task_desc,task.script,task.data_status,task.create_user_id,task.create_time,
task.update_user_id,task.update_time,task.tree_id,task.flow_header,
task.flow_json, task.version, task.publish_version, task.is_gziped, tree.name
FROM dmp_develop_task task left join dmp_navigation_tree tree ON task.tree_id=tree.id
where task.data_status ='1' and task.TREE_ID = #{treeId}
</select>
<!--数据运维-数据开发任务列表分页查询-->
......@@ -181,9 +188,9 @@
SELECT
task.id,task.task_type,task.type,task.schedule_type,task.is_submit,task.task_desc,task.script,task.data_status,task.create_user_id,task.create_time,
task.update_user_id,task.update_time,task.tree_id,task.flow_header,
task.flow_json, task.version,task.is_gziped
task.flow_json, task.version, task.publish_version, task.is_gziped, tree.name
FROM dmp_develop_task task left join dmp_navigation_tree tree ON task.tree_id=tree.id
WHERE 1=1
WHERE 1=1 AND tree.data_status = '1'
<if test="taskType != null">AND task.task_type = #{taskType}</if>
<if test="type != null">AND task.type = #{type}</if>
<if test="scheduleType != null">AND task.schedule_type = #{scheduleType}</if>
......@@ -196,6 +203,7 @@
<if test="flowHeader != null">AND task.flow_header = #{flowHeader}</if>
<if test="flowJson != null">AND task.flow_json = #{flowJson}</if>
<if test="version != null">AND task.version = #{version}</if>
<if test="publishVersion != null">AND task.publish_version = #{publishVersion}</if>
<if test="gziped != null">AND task.is_gziped = #{gziped}</if>
<if test="projectId != null">AND tree.project_id = #{projectId}</if>
</select>
......@@ -218,7 +226,12 @@
<!-- 主键获取对象 -->
<select id="get" parameterType="java.lang.Long" resultMap="DmpDevelopTaskResultMap">
SELECT <include refid="FIND_ALL_COLUMN" /> FROM dmp_develop_task WHERE tree_id = #{id}
SELECT
task.id,task.task_type,task.type,task.schedule_type,task.is_submit,task.task_desc,task.script,task.data_status,task.create_user_id,task.create_time,
task.update_user_id,task.update_time,task.tree_id,task.flow_header,
task.flow_json, task.version, task.publish_version, task.is_gziped, tree.name, tree.project_id
FROM dmp_develop_task task left join dmp_navigation_tree tree ON task.tree_id=tree.id
WHERE task.tree_id = #{id}
</select>
<select id="selectTaskById" resultType="com.jz.dmp.modules.model.DmpDevelopTask">
......
......@@ -265,6 +265,16 @@
<delete id="deleteById">
delete from dmp_navigation_tree where ID = #{id}
</delete>
<!--假删除任务树-->
<delete id="deleteByTreeId" parameterType="java.util.HashMap">
update dmp_navigation_tree
<trim prefix="SET" suffixOverrides=",">
<if test="dataStatus != null">
data_status = #{dataStatus},
</if>
</trim>
where id = #{ids}
</delete>
<select id="countTreeByName" parameterType="com.jz.dmp.modules.model.DmpNavigationTree" resultType="java.lang.Integer">
select
......
......@@ -4,6 +4,7 @@
<resultMap type="com.jz.dmp.modules.model.DmpRealtimeSyncInfo" id="DmpRealtimeSyncInfoMap">
<result property="id" column="id" jdbcType="INTEGER"/>
<result property="treeId" column="tree_id" jdbcType="VARCHAR"/>
<result property="srcDatasourceId" column="src_datasource_id" jdbcType="INTEGER"/>
<result property="targetDatasourceId" column="target_datasource_id" jdbcType="INTEGER"/>
<result property="srcTableName" column="src_table_name" jdbcType="VARCHAR"/>
......@@ -37,7 +38,7 @@
<!--查询单个-->
<select id="queryById" resultMap="DmpRealtimeSyncInfoMap">
select
id, src_datasource_id, target_datasource_id, src_table_name, target_table_name, type, connector_job_id, connector_json_data
id, tree_id, src_datasource_id, target_datasource_id, src_table_name, target_table_name, type, connector_job_id, connector_json_data
, src_topic_name, project_id, parent_id, desensitization_field, arithmetic, pk_name, source_type_name, target_type_name
, src_database_type, src_database_name, connector_url, target_database_type, target_database_name, src_datasource_name
, target_datasource_name, store_type, status, create_time, update_time, cre_person, upt_person
......
......@@ -83,47 +83,47 @@
"params": {
"version": "1.0", //版本
"treeId": 669,
//"parentId": "509",
"mode": "0",
"projectId": "31",
"taskId":"", //任务id
"taskName": "dmp_demo_dmp_azkaban_exector_server_config", //任务名称
"scripts": {
"setting": {
"extract": "incremental", //增量/全量
"extractExpression": "where 1=1", //增量表达式
"targetInsertMergeOverwrite": "insert", //插入合并重写
//"extract": "incremental", //增量/全量
//"extractExpression": "where 1=1", //增量表达式 ,数据过滤
//"targetInsertMergeOverwrite": "insert", //插入合并重写
"ftColumn": "分桶字段",
"ftCount": "分桶个数",
"separateMax": "分桶字段最大值",
"separateMin": "分桶字段最小值",
//"primaryKey": "主键",
//"partition": "分区",
"postImportStatement": "导入后语句",
"preImportStatement": "导入前语句",
"errorLimitRecord": "错误记录数超过",
"maxConcurrency": "最大并发数",
//"syncRate": "同步速率",
"executorMemory":"1", //分配任务内存
"executorCores":"1", //单executor的cpu数
"totalExecutorCores":"1", //总executor的cpu数
"driverMemory":"",//分配任务内存
"executorCore":"1", //单executor的cpu数
"driverCore":"1",//单executor的cpu数
"fieldMapping":""//字段映射关系
},
"reader": {
"dbConnection": "mysql_dmp_demo_test", //来源名称
"fileType": "", //文件类型
"registerTableName": "dmp_azkaban_exector_server_config",
"sourceHdfsPath": "", //HDFS存储目录
"sourceHdfsFile": "",
"sourceFtpDir": "", //文件所在目录
"sourceFtpFile": "", //文件名
"sourceSkipFtpFile": "", //没有数据文件是否跳过
"fileType": "", //文件类型
"sourceCsvDelimiter": "", //分隔符
"sourceCsvHeader": "", //是否含有表头
"sourceCsvCharset": "", //字符集编码
"sourceCsvHeader": "", //是否含有表头
"null值":"",
"compressedFormat":"", //压缩格式
"sourceFtpFile": "", //文件名
"sourceSkipFtpFile": "", //没有数据文件是否跳过
"sourceCsvQuote": "",
"sourceFtpLoadDate": "", //加载数据日期
"registerTableName": "dmp_azkaban_exector_server_config",
"dayByDay": "false", //day_by_day
"syncopationKey":"", //切分键
"extractExpression": "1=1 and id=1", //增量表达式 ,数据过滤
"readHiveMethod ":"",//读取hive方法
"querySql":"",//hive数据查询sql
"column": [
{
"name": "host",
......@@ -146,12 +146,22 @@
"writer": {
"targetDbConnection": "mysql_dmp_demo",
"targetTable": "dmp_azkaban_exector_server_config",
"targetFtpDir": "",
"targetFtpFile": "",
"targetCsvDelimiter": "",
"targetFtpDir": "",//文件所在目录
"targetFtpFile": "",//文件名称
"targetCsvDelimiter": "", //列分隔符
"targetCsvCharset": "",//字符集编码
"fileType": "", //文件类型
"datatimeFormat":"",//时间格式
"null值":"",//表示有null值得字符串
"prefixConflict":"",//前缀冲突
"markCompletionFile":"",//标记完成文件
"targetCsvHeader": "",
"targetCsvCharset": "",
"targetInsertMergeOverwrite": "insert",
"postImportStatement": "",//导入后语句
"preImportStatement": "",//导入前语句
"primaryKeyConflict":"",//主键冲突
"partitionConflict":"",//分区冲突
"writeMode":"",//hive写入模式
"targetInsertMergeOverwrite": "insert", //数据插入方式
"column": [
{
"name": "host",
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment