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

添加字段和表

master
Cui YongXing 2024-09-02 19:27:49 +08:00
parent d52b2f8768
commit 89fc58a0e7
4 changed files with 57 additions and 59 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/service/TaskInfoService.java
View File

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

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

@ -8,7 +8,7 @@ import com.muyu.common.domian.req.TaskInputListReq;
public interface TaskInputService extends IService<TaskInput> {
Page<TaskInput> findList(TaskInputListReq req);
String findByFieName(Long taskId);
TaskInput selectByNodeId(String oneNodeId);

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

@ -63,4 +63,59 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
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;
}
}

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

@ -46,63 +46,6 @@ public class TaskInputServiceImpl extends ServiceImpl<TaskInputMapper, TaskInput
return this.page(page, wrapper);
}
@Override
public String findByFieName(Long taskId) {
QueryWrapper<TaskInput> wrapper = new QueryWrapper<>();
wrapper.eq("task_id", taskId);
HashSet<Long> longs = new HashSet<>();
List<TaskInput> list = taskInputMapper.selectList(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 = taskInputMapper.selectByNodeId(oneNodeId);
String tableName = taskInput.getTableName();
String tableAsName = taskInput.getTableAsName();
String twoNodeId = nodeJoint.getTwoNodeId();
TaskInput taskInputTwo = taskInputMapper.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 = taskInputMapper.selectByTaskId(taskId);
String tableName = taskInput.getTableName();
String tableAsName = taskInput.getTableAsName();
joint=" "+tableName+" "+tableAsName;
}
Long basicId = taskInputMapper.selectByBasicId(taskId);
fieName = fieName.substring(1);
String sql="select "+fieName + " from "+joint;
System.out.println("select "+fieName + " from "+joint);
Result<List<List<DataValue>>> tableValue = datasourceFeign.findTableValue(basicId, sql);
System.out.println(tableValue);
return null;
}
@Override
public TaskInput selectByNodeId(String oneNodeId) {
return taskInputMapper.selectByNodeId(oneNodeId);