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

添加字段和表

master
Cui YongXing 2024-09-02 21:20:37 +08:00
parent 5f812afbb7
commit b1b7efd3e7
3 changed files with 106 additions and 12 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){
taskInfoService.findByFieName(taskId);
taskInfoService.findByFieName2(taskId);
return Result.success();
}

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

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

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

@ -24,6 +24,7 @@ import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import javax.annotation.Resource;
import java.lang.reflect.Constructor;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@ -69,9 +70,87 @@ 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 = "";
// HashMap<String, Long> hashMap = new HashMap<>();
// for (int i = 0; i < list.size(); i++) {
// fieName += "," + list.get(i).getTableAsFieId();
// Long databaseId = list.get(i).getDatabaseId();
// longs.add(databaseId);
// String tableFieId = list.get(i).getTableFieId();
// Long fieIdAsEngineId = list.get(i).getFieIdAsEngineId();
// hashMap.put(tableFieId,fieIdAsEngineId);
// }
// 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);
// QueryWrapper<TaskOutput> queryWrapper = new QueryWrapper<>();
// queryWrapper.eq("task_id", taskId);
// List<TaskOutput> list1 = taskOutputService.list(queryWrapper);
// HashMap<String, String> map = new HashMap<>();
// Long tableId=null;
// Long newBasicId=null;
// for (TaskOutput taskOutput : list1) {
// tableId=taskOutput.getTableId();
// newBasicId=taskOutput.getBasicId();
// String lastFieName = taskOutput.getLastFieName();
// String newFieName = taskOutput.getNewFieName();
// map.put(lastFieName, newFieName);
// }
// List<List<DataValue>> tableValue = tableValueResult.getData();
// for (List<DataValue> dataValues : tableValue) {
// for (DataValue dataValue : dataValues) {
// String key = dataValue.getKey();
// String newKey = map.get(key);
// dataValue.setKey(newKey);
// }
// }
// return "执行成功";
// }
@Override
public String findByFieName(Long taskId) {
public String findByFieName2(Long taskId) {
QueryWrapper<TaskInput> wrapper = new QueryWrapper<>();
wrapper.eq("task_id", taskId);
HashSet<Long> longs = new HashSet<>();
@ -79,19 +158,22 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
if (CollectionUtils.isEmpty(list)) {
return "没有选择字段";
}
HashMap<String, String> tableNameMap = new HashMap<>();
String fieName = "";
HashMap<String, Long> hashMap = new HashMap<>();
for (int i = 0; i < list.size(); i++) {
fieName += "," + list.get(i).getTableAsFieId();
Long databaseId = list.get(i).getDatabaseId();
longs.add(databaseId);
String tableFieId = list.get(i).getTableFieId();
String tableAsFieId = list.get(i).getTableAsFieId();
Long fieIdAsEngineId = list.get(i).getFieIdAsEngineId();
hashMap.put(tableFieId,fieIdAsEngineId);
tableNameMap.put(tableFieId,tableAsFieId);
}
if (longs.size() > 1) {
return "你选择的不是同一个数据库";
}
QueryWrapper<NodeJoint> jointQueryWrapper = new QueryWrapper<>();
jointQueryWrapper.eq("task_id", taskId);
List<NodeJoint> jointList = nodeJointService.list(jointQueryWrapper);
@ -119,26 +201,36 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
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);
QueryWrapper<TaskOutput> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("task_id", taskId);
List<TaskOutput> list1 = taskOutputService.list(queryWrapper);
HashMap<String, String> map = new HashMap<>();
Long tableId=null;
Long newBasicId=null;
for (TaskOutput taskOutput : list1) {
tableId=taskOutput.getTableId();
newBasicId=taskOutput.getBasicId();
String lastFieName = taskOutput.getLastFieName();
String newFieName = taskOutput.getNewFieName();
map.put(lastFieName, newFieName);
fieName += "," + tableNameMap.get(lastFieName);
}
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) {
for (DataValue dataValue : dataValues) {
String key = dataValue.getKey();
String newKey = map.get(key);
dataValue.setKey(newKey);
}
}
return "执行成功";
}
}