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

添加字段和表

master
Cui YongXing 2024-09-01 14:37:36 +08:00
parent 823631867d
commit a113cda05f
17 changed files with 134 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cloud-task-common/pom.xml
View File

@ -21,5 +21,9 @@
<groupId>com.muyu</groupId>
<artifactId>cloud-common-core</artifactId>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-datasources-common</artifactId>
</dependency>
</dependencies>
</project>

2
cloud-task-common/src/main/java/com/muyu/common/domian/NodeJoint.java
View File

@ -25,7 +25,7 @@ public class NodeJoint {
*/
private Long taskId;
/**
* 任务id
* 节点id
*/
private Long nodeId;
/**

2
cloud-task-common/src/main/java/com/muyu/common/domian/TaskInput.java
View File

@ -46,7 +46,7 @@ public class TaskInput extends BaseEntity {
/**
* 数据库id
*/
private String databaseId;
private Long databaseId;
/**
* 表名称

2
cloud-task-common/src/main/java/com/muyu/common/domian/req/InputAddReq.java
View File

@ -33,7 +33,7 @@ public class InputAddReq {
/**
* 数据库id
*/
private String databaseId;
private Long databaseId;
/**
* 表名称

2
cloud-task-common/src/main/java/com/muyu/common/domian/req/TaskInputUpdReq.java
View File

@ -30,7 +30,7 @@ public class TaskInputUpdReq {
/**
* 数据库id
*/
private String databaseId;
private Long databaseId;
/**
* 表名称

2
cloud-task-common/src/main/java/com/muyu/common/domian/resp/TaskInputResp.java
View File

@ -35,7 +35,7 @@ public class TaskInputResp {
/**
* 数据库id
*/
private String databaseId;
private Long databaseId;
/**
* 表名称

6
cloud-task-remote/pom.xml
View File

@ -16,5 +16,11 @@
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-task-common</artifactId>
</dependency>
</dependencies>
</project>

7
cloud-task-remote/src/main/java/com/muyu/Main.java
View File

@ -1,7 +0,0 @@
package com.muyu;
public class Main {
public static void main(String[] args) {
System.out.println("Hello world!");
}
}

1
cloud-task-remote/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports 100644
View File

@ -0,0 +1 @@
com.muyu.task.server.feign.Factory.DatasourceFeignFactory

10
cloud-task-server/pom.xml
View File

@ -90,7 +90,15 @@
<groupId>com.muyu</groupId>
<artifactId>cloud-common-nacos-api</artifactId>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-task-remote</artifactId>
<version>1.0.0</version>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-datasources-common</artifactId>
</dependency>
</dependencies>

1
cloud-task-server/src/main/java/com/muyu/task/server/TaskApplication.java
View File

@ -5,6 +5,7 @@ import com.muyu.common.security.annotation.EnableMyFeignClients;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@EnableCustomConfig
@EnableMyFeignClients
@SpringBootApplication

22
cloud-task-server/src/main/java/com/muyu/task/server/feign/DatasourceFeign.java 100644
View File

@ -0,0 +1,22 @@
package com.muyu.task.server.feign;
import com.muyu.common.domain.DataValue;
import com.muyu.task.server.feign.Factory.DatasourceFeignFactory;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import com.muyu.common.core.domain.Result;
import java.util.List;
/**
* @author Administrator
*/
@EnableFeignClients
@FeignClient(contextId = "datasourceFeign", name = "cloud-etl-datasources",fallbackFactory = DatasourceFeignFactory.class )
public interface DatasourceFeign {
@PostMapping("/dataValue/findTableValue")
public Result<List<List<DataValue>>> findTableValue(@RequestParam("basicId") Long basicId, @RequestParam("sql") String sql);
}

22
cloud-task-server/src/main/java/com/muyu/task/server/feign/Factory/DatasourceFeignFactory.java 100644
View File

@ -0,0 +1,22 @@
package com.muyu.task.server.feign.Factory;
import com.muyu.common.core.domain.Result;
import com.muyu.task.server.feign.DatasourceFeign;
import lombok.extern.log4j.Log4j2;
import org.springframework.cloud.openfeign.FallbackFactory;
import org.springframework.stereotype.Component;
@Log4j2
@Component
public class DatasourceFeignFactory implements FallbackFactory<DatasourceFeign> {
@Override
public DatasourceFeign create(Throwable e) {
return new DatasourceFeign() {
@Override
public Result findTableValue(Long basicId, String sql) {
log.info(e);
return Result.error("网络开小差......");
}
};
}
}

2
cloud-task-server/src/main/java/com/muyu/task/server/mapper/TaskInputMapper.java
View File

@ -8,5 +8,7 @@ import org.apache.ibatis.annotations.Mapper;
public interface TaskInputMapper extends BaseMapper<TaskInput> {
TaskInput selectByNodeId(String oneNodeId);
TaskInput selectByTaskId(Long taskId);
}

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

@ -6,9 +6,11 @@ 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.TaskInput;
import com.muyu.common.domian.req.TaskInputListReq;
import com.muyu.task.server.feign.DatasourceFeign;
import com.muyu.task.server.mapper.TaskInputMapper;
import com.muyu.task.server.service.NodeJointService;
import com.muyu.task.server.service.NodeTableService;
@ -17,6 +19,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import java.util.HashSet;
import java.util.List;
@Service
@ -30,6 +33,10 @@ public class TaskInputServiceImpl extends ServiceImpl<TaskInputMapper, TaskInput
@Autowired
private NodeJointService nodeJointService;
@Autowired
private DatasourceFeign datasourceFeign;
@Override
public Page<TaskInput> findList(TaskInputListReq req) {
LambdaQueryWrapper<TaskInput> wrapper = new LambdaQueryWrapper<>();
@ -40,30 +47,57 @@ public class TaskInputServiceImpl extends ServiceImpl<TaskInputMapper, TaskInput
}
@Override
public String findByFieName(Long taskId) {
QueryWrapper<TaskInput> wrapper = new QueryWrapper<>();
wrapper.eq("task_id",taskId);
wrapper.eq("task_id", taskId);
HashSet<Long> longs = new HashSet<>();
List<TaskInput> list = taskInputMapper.selectList(wrapper);
if (CollectionUtils.isEmpty(list)){
if (CollectionUtils.isEmpty(list)) {
return "没有选择字段";
}
String fieName="";
String fieName = "";
for (int i = 0; i < list.size(); i++) {
fieName+= ","+list.get(i).getTableAsFieId();
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);
jointQueryWrapper.eq("task_id", taskId);
List<NodeJoint> jointList = nodeJointService.list(jointQueryWrapper);
for (NodeJoint nodeJoint : jointList) {
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;
}
TaskInput taskInput = taskInputMapper.selectByTaskId(taskId);
Long databaseId = taskInput.getDatabaseId();
fieName = fieName.substring(1);
// tableService.selectTableName(taskId);
String sql="select "+fieName + " form "+joint;
System.out.println("select "+fieName + " from "+joint);
Result<List<List<DataValue>>> tableValue = datasourceFeign.findTableValue(databaseId, sql);
System.out.println(tableValue);
return null;
}
}

7
cloud-task-server/src/main/resources/mapper/TaskInfoMapper.xml
View File

@ -1,7 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper
PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.muyu.task.server.mapper.TaskInfoMapper">
</mapper>

17
cloud-task-server/src/main/resources/mapper/TaskInputMapper.xml 100644
View File

@ -0,0 +1,17 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper
PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.muyu.task.server.mapper.TaskInputMapper">
<select id="selectByNodeId" resultType="com.muyu.common.domian.TaskInput">
SELECT DISTINCT table_name,table_as_name
FROM etl_task_input
where node_id =${nodeId}
</select>
<select id="selectByTaskId" resultType="com.muyu.common.domian.TaskInput">
SELECT DISTINCT database_id
FROM etl_task_input
where task_id =${taskId}
</select>
</mapper>