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

fix(): 修改数据接入数据,同步数据不同步

dev
chao 2024-04-23 15:30:34 +08:00
parent bd68e9030f
commit 2101257954
2 changed files with 106 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
etl-modules/etl-modules-data-source/etl-modules-data-source-common/src/main/java/com/etl/data/structure/domain/AssetStructure.java
View File

@ -52,7 +52,7 @@ public class AssetStructure extends BaseEntity {
*/
private String dataSourceDatabaseName;
public static AssetStructure dataSourceBuilder(Long id, DataSource dataSource) {
public static AssetStructure dataSourceSaveBuilder(Long id, DataSource dataSource) {
return AssetStructure.builder()
.dataSourceSystemId(id)
.dataSourceName(dataSource.getDataSourceName())
@ -60,4 +60,14 @@ public class AssetStructure extends BaseEntity {
.dataSourceDatabaseName(dataSource.getDataSourceDatabaseName())
.build();
}
public static AssetStructure dataSourceUpdateBuilder(AssetStructure assetStructure, Long id, DataSource dataSource) {
return AssetStructure.builder()
.id(assetStructure.getId())
.dataSourceSystemId(id)
.dataSourceName(dataSource.getDataSourceName())
.dataSourceSystemName(dataSource.getDataSourceSystemName())
.dataSourceDatabaseName(dataSource.getDataSourceDatabaseName())
.build();
}
}

96
etl-modules/etl-modules-data-source/etl-modules-data-source-system/src/main/java/com/etl/data/source/service/impl/DataSourceServiceImpl.java
View File

@ -1,6 +1,7 @@
package com.etl.data.source.service.impl;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.etl.common.core.utils.DateUtils;
@ -210,7 +211,8 @@ public class DataSourceServiceImpl extends ServiceImpl<DataSourceMapper, DataSou
if (b) {
// 如果不存在就给他添加资产结构表
if (assetStructure == null) {
AssetStructure entity = AssetStructure.dataSourceBuilder(id, dataSource);
// 构建实体
AssetStructure entity = AssetStructure.dataSourceSaveBuilder(id, dataSource);
assetStructureService.save(entity);
String jdbcUrl = "";
String driveClass = "";
@ -291,8 +293,100 @@ public class DataSourceServiceImpl extends ServiceImpl<DataSourceMapper, DataSou
System.out.println(e);
return false;
}
} else { // 如果他存在就给他修改
AssetStructure entity = AssetStructure.dataSourceUpdateBuilder(assetStructure,id, dataSource);
// 修改资产结构
assetStructureService.update(
entity,new LambdaUpdateWrapper<AssetStructure>()
.eq(AssetStructure::getId,entity.getId())
);
// 删除资产结构表信息
assetStructureTableService.remove(
new LambdaQueryWrapper<AssetStructureTable>()
.eq(AssetStructureTable::getAssetStructureId,entity.getId())
);
String jdbcUrl = "";
String driveClass = "";
DataType dataType = dataTypeService.getOne(
new LambdaQueryWrapper<DataType>()
.eq(DataType::getId, dataSource.getTypeId())
);
try {
List<AssetStructureTable> assetStructureTableList = new ArrayList<>();
if (dataType.getDriverManager() != null && dataType.getJdbcPre() != null) {
if ("mysql".equals(dataType.getDataType())) {
driveClass = dataType.getDriverManager();
jdbcUrl = dataType.getJdbcPre() + dataSource.getDataSourceIp() + ":" + dataSource.getDataSourcePort() + "/" + dataSource.getDataSourceDatabaseName() + "?" + dataSource.getAdditionalConfiguration();
// 加载数据库驱动
Class.forName(driveClass);
// 连接数据库
Connection conn = DriverManager.getConnection(jdbcUrl, dataSource.getDataSourceUsername(), dataSource.getDataSourcePassword());
Statement st = conn.createStatement();
ResultSet rs = st.executeQuery("select * from information_schema.tables where TABLE_SCHEMA = " + "'" + dataSource.getDataSourceDatabaseName() + "'");
while (rs.next()) {
// 获取表名
String tableName = rs.getString("TABLE_NAME");
// 表注释
String tableNameAnnotation = rs.getString("TABLE_COMMENT");
// 添加数据
// 表数据数量
assetStructureTableList.add(
new AssetStructureTable() {{
setAssetStructureId(entity.getId());
setTableName(tableName);
setTableNameAnnotation(tableNameAnnotation);
}}
);
}
assetStructureTableList.stream().forEach(assetStructureTable -> {
try {
ResultSet rs2 = st.executeQuery("select count(*) as countNum from " + assetStructureTable.getTableName());
while (rs2.next()){
assetStructureTable.setTableDataCount(rs2.getLong("countNum"));
}
rs2.close();
} catch (SQLException e) {
}
});
conn.close();
st.close();
rs.close();
// 批量插入
assetStructureTableService.saveBatch(assetStructureTableList);
}
if ("oracle".equals(dataType.getDataType())) {
driveClass = dataType.getDriverManager();
jdbcUrl = dataType.getJdbcPre() + dataSource.getDataSourceIp() + ":" + dataSource.getDataSourcePort() + ":" + dataSource.getDataSourceDatabaseName();
}
if ("sqlserver".equals(dataType.getDataType())) {
driveClass = dataType.getDriverManager();
jdbcUrl = dataType.getJdbcPre() + dataSource.getDataSourceIp() + ":" + dataSource.getDataSourcePort() + ";databaseName=" + dataSource.getDataSourceDatabaseName();
}
} else {
// redis
//连接指定的redis
Jedis jedis = new Jedis(dataSource.getDataSourceIp(), Integer.valueOf(dataSource.getDataSourcePort()));
//如果有密码则需要下面这一行
if (dataSource.getDataSourcePassword() != null && !dataSource.getDataSourcePassword().equals("")) {
jedis.auth(dataSource.getDataSourcePassword());
}
//查看服务是否运行,运行正常的话返回一个PONG，否则返回一个连接错误
try {
jedis.ping();
} catch (Exception e) {
}
}
} catch (Exception e) {
System.out.println(e);
return false;
}
}
} else {
// 如果连接失败并存在就给他删除