etl
/
cloud-etl-task
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

添加字段和表

master
Cui YongXing 2024-09-02 18:54:46 +08:00
parent 57f4f2a59c
commit f3d74d4dd2
5 changed files with 100 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cloud-task-server/src/main/java/com/muyu/task/server/controller/TaskInfoController.java
View File

@ -118,7 +118,7 @@ public class TaskInfoController {
@PostMapping("/execute")
public Result execute(@RequestParam("taskId") Long taskId){
taskInputService.findByFieName(taskId);
taskInfoService.findByFieName(taskId);
return Result.success();
}

2
cloud-task-server/src/main/java/com/muyu/task/server/service/TaskInfoService.java
View File

@ -12,4 +12,6 @@ import com.muyu.common.domian.resp.TaskInfoResp;
public interface TaskInfoService extends IService<TaskInfo> {
Page<TaskInfoResp> selectList(TaskInfoListReq req);
String findByFieName(Long taskId);
}

8
cloud-task-server/src/main/java/com/muyu/task/server/service/TaskInputService.java
View File

@ -5,10 +5,14 @@ import com.baomidou.mybatisplus.extension.service.IService;
import com.muyu.common.domian.TaskInput;
import com.muyu.common.domian.req.TaskInputListReq;
import java.util.List;
public interface TaskInputService extends IService<TaskInput> {
Page<TaskInput> findList(TaskInputListReq req);
String findByFieName(Long taskId);
TaskInput selectByNodeId(String oneNodeId);
TaskInput selectByTaskId(Long taskId);
Long selectByBasicId(Long taskId);
}

77
cloud-task-server/src/main/java/com/muyu/task/server/service/impl/TaskInfoServiceImpl.java
View File

@ -1,17 +1,28 @@
package com.muyu.task.server.service.impl;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.muyu.common.core.domain.Result;
import com.muyu.common.core.utils.StringUtils;
import com.muyu.common.domain.DataValue;
import com.muyu.common.domian.NodeJoint;
import com.muyu.common.domian.TaskInfo;
import com.muyu.common.domian.TaskInput;
import com.muyu.common.domian.req.TaskInfoListReq;
import com.muyu.common.domian.resp.TaskInfoResp;
import com.muyu.remote.feign.DatasourceFeign;
import com.muyu.task.server.mapper.TaskInfoMapper;
import com.muyu.task.server.service.NodeJointService;
import com.muyu.task.server.service.TaskInfoService;
import com.muyu.task.server.service.TaskInputService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import javax.annotation.Resource;
import java.util.HashSet;
import java.util.List;
/**
@ -22,12 +33,20 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
@Resource
private TaskInfoMapper taskInfoMapper;
@Autowired
private TaskInputService taskInputService;
@Autowired
private NodeJointService nodeJointService;
@Autowired
private DatasourceFeign datasourceFeign;
@Override
public Page<TaskInfoResp> selectList(TaskInfoListReq req) {
LambdaQueryWrapper<TaskInfo> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.like(StringUtils.isNotEmpty(req.getName()), TaskInfo::getName, req.getName());
queryWrapper.eq(req.getWeight()!=null, TaskInfo::getWeight, req.getWeight());
queryWrapper.eq(req.getWeight() != null, TaskInfo::getWeight, req.getWeight());
queryWrapper.eq(StringUtils.isNotEmpty(req.getProcessStatus()), TaskInfo::getProcessStatus, req.getProcessStatus());
Page<TaskInfo> page = new Page<>(req.getPageNum(), req.getPageSize());
Page<TaskInfo> taskInfoPage = taskInfoMapper.selectPage(page, queryWrapper);
@ -43,4 +62,60 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
respPage.setPages(pages);
return respPage;
}
@Override
public String findByFieName(Long taskId) {
QueryWrapper<TaskInput> wrapper = new QueryWrapper<>();
wrapper.eq("task_id", taskId);
HashSet<Long> longs = new HashSet<>();
List<TaskInput> list = taskInputService.list(wrapper);
if (CollectionUtils.isEmpty(list)) {
return "没有选择字段";
}
String fieName = "";
for (int i = 0; i < list.size(); i++) {
fieName += "," + list.get(i).getTableAsFieId();
Long databaseId = list.get(i).getDatabaseId();
longs.add(databaseId);
}
if (longs.size() > 1) {
return "你选择的不是同一个数据库";
}
QueryWrapper<NodeJoint> jointQueryWrapper = new QueryWrapper<>();
jointQueryWrapper.eq("task_id", taskId);
List<NodeJoint> jointList = nodeJointService.list(jointQueryWrapper);
String joint = "";
if (!CollectionUtils.isEmpty(jointList)) {
for (NodeJoint nodeJoint : jointList) {
String oneNodeId = nodeJoint.getOneNodeId();
TaskInput taskInput = taskInputService.selectByNodeId(oneNodeId);
String tableName = taskInput.getTableName();
String tableAsName = taskInput.getTableAsName();
String twoNodeId = nodeJoint.getTwoNodeId();
TaskInput taskInputTwo = taskInputService.selectByNodeId(twoNodeId);
String tableNameTwo = taskInputTwo.getTableName();
String tableAsNameTwo = taskInputTwo.getTableAsName();
joint += " " + tableName + " " + " " + tableAsName + " " + nodeJoint.getJoint()
+ " " + tableNameTwo + " " + " " + tableAsNameTwo + " on " + nodeJoint.getOneFie()
+ " = " + nodeJoint.getTwoFie();
}
}else {
TaskInput taskInput = taskInputService.selectByTaskId(taskId);
String tableName = taskInput.getTableName();
String tableAsName = taskInput.getTableAsName();
joint=" "+ tableName+" "+tableAsName;
}
Long basicId = taskInputService.selectByBasicId(taskId);
fieName = fieName.substring(1);
String sql = " SELECT " + fieName + " FROM "+ joint;
System.out.println(sql);
Result<List<List<DataValue>>> tableValueResult = datasourceFeign.findTableValue(basicId, sql);
List<List<DataValue>> tableValue = tableValueResult.getData();
for (List<DataValue> dataValues : tableValue) {
System.out.println(dataValues);
}
return null;
}
}

15
cloud-task-server/src/main/java/com/muyu/task/server/service/impl/TaskInputServiceImpl.java
View File

@ -102,4 +102,19 @@ public class TaskInputServiceImpl extends ServiceImpl<TaskInputMapper, TaskInput
return null;
}
@Override
public TaskInput selectByNodeId(String oneNodeId) {
return taskInputMapper.selectByNodeId(oneNodeId);
}
@Override
public TaskInput selectByTaskId(Long taskId) {
return taskInputMapper.selectByTaskId(taskId);
}
@Override
public Long selectByBasicId(Long taskId) {
return taskInputMapper.selectByBasicId(taskId);
}
}