package com.highdatas.mdm.service.impl;
|
|
import com.baomidou.mybatisplus.mapper.EntityWrapper;
|
import com.highdatas.mdm.entity.*;
|
import com.highdatas.mdm.mapper.SysAssembleMapper;
|
import com.highdatas.mdm.mapper.TableInfoMapper;
|
import com.highdatas.mdm.pojo.*;
|
import com.highdatas.mdm.pojo.kettle.BigDataDataSourceInfo;
|
import com.highdatas.mdm.pojo.kettle.DataSourceInfo;
|
import com.highdatas.mdm.pojo.kettle.UnBigDataDataSourceInfo;
|
import com.highdatas.mdm.service.*;
|
import com.baomidou.mybatisplus.service.impl.ServiceImpl;
|
import com.highdatas.mdm.util.Constant;
|
import com.highdatas.mdm.util.ContentBuilder;
|
import com.highdatas.mdm.util.DbUtils;
|
import com.highdatas.mdm.util.RuleClient;
|
import lombok.Data;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.commons.lang3.StringUtils;
|
import org.apache.ibatis.session.SqlSession;
|
import org.apache.regexp.RE;
|
import org.jcodings.util.Hash;
|
import org.mybatis.spring.SqlSessionTemplate;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.stereotype.Service;
|
import org.springframework.transaction.annotation.Transactional;
|
|
import javax.annotation.Resource;
|
import javax.servlet.http.HttpSession;
|
import java.io.ByteArrayInputStream;
|
import java.io.InputStream;
|
import java.io.UnsupportedEncodingException;
|
import java.sql.*;
|
import java.text.MessageFormat;
|
import java.util.*;
|
import java.util.Date;
|
import java.util.stream.Collectors;
|
|
/**
|
* <p>
|
* 服务实现类
|
* </p>
|
*
|
* @author kimi
|
* @since 2020-02-20
|
*/
|
@Slf4j
|
@Service
|
public class SysAssembleServiceImpl extends ServiceImpl<SysAssembleMapper, SysAssemble> implements ISysAssembleService {
|
@Autowired
|
ISysAssembleRuleService ruleService;
|
@Autowired
|
ISysAssembleDbService dbService;
|
@Autowired
|
ISysAssembleApiService apiService;
|
@Autowired
|
ISysAssembleDbTableService tableService;
|
@Autowired
|
ISysAssembleDbFieldService fieldService;
|
@Autowired
|
ISysAssembleParamsService paramsService;
|
@Autowired
|
ISysAssembleLogService logService;
|
@Autowired
|
IMenuMappingService menuMappingService;
|
@Autowired
|
ActivitiService activitiService;
|
@Autowired
|
IMaintainService maintainService;
|
@Autowired
|
IFlowsService flowsService;
|
@Resource
|
SqlSessionTemplate sqlSessionTemplate;
|
@Autowired
|
TableInfoMapper tableInfoMapper;
|
@Autowired
|
UnBigDataDataSourceInfo unBigDataDataSourceInfo;
|
@Autowired
|
BigDataDataSourceInfo bigDataDataSourceInfo;
|
@Autowired
|
MasterDataService masterDataService;
|
@Autowired
|
ITUserService userService;
|
@Autowired
|
RuleClient ruleClient;
|
@Autowired
|
IMasterModifiedService masterModifiedService;
|
|
@Override
|
public Result run(String id) {
|
if (StringUtils.isEmpty(id)) {
|
return Result.error(CodeMsg.ERROR_PARAMS_NOT_MATHED);
|
}
|
|
SysAssemble assemble = selectById(id);
|
if (assemble == null) {
|
return Result.error(CodeMsg.ERROR_PARAMS_NOT_MATHED);
|
}
|
assemble.setPreTime(new Date());
|
assemble.setPreStatus(SysAssembleRunStatus.working);
|
assemble.updateById();
|
|
if (!assemble.getStatus().equals(SysAssembleStatus.working)) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreCnt(0);
|
assemble.setPreMsg("当前任务不在工作中,状态:" + assemble.getStatus());
|
assemble.updateById();
|
return Result.error(new CodeMsg(6009,"当前任务不在工作中,状态:" + assemble.getStatus()));
|
|
}
|
|
String flowId = assemble.getFlowId();
|
if (!StringUtils.isEmpty(flowId)){
|
Flows flows = flowsService.selectById(flowId);
|
if (flows != null) {
|
ActivitiStatus status = flows.getStatus();
|
if (!status.equals(ActivitiStatus.close) && !status.equals(ActivitiStatus.open)) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreCnt(0);
|
assemble.setPreMsg("当前有流程正在运行,暂时无法汇集下次数据:" + status.toString());
|
assemble.updateById();
|
return Result.error(new CodeMsg(6009,"当前有流程正在运行,暂时无法汇集下次数据:" + status.toString()));
|
|
}
|
}
|
}
|
|
|
Boolean bigData = assemble.getBigdata();
|
|
//1 load from db
|
try{
|
Result result = runByDb(assemble);
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.updateById();
|
return result;
|
}
|
//2 load from api
|
result = runByApi(id);
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.updateById();
|
return result;
|
}
|
//3check temp table;
|
result = checkTempTable(assemble);
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreMsg("检测主数据临时表出现错误");
|
assemble.updateById();
|
return result;
|
}
|
|
//4 purge data
|
String purgeSql = assemble.getPurgeSql();
|
purgeSql = DbUtils.replaceEscape(purgeSql);
|
result = purgeData(purgeSql, assemble.getId(), bigData);
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreMsg("运行清洗sql出现错误: "+purgeSql );
|
assemble.updateById();
|
return result;
|
}
|
|
//5 check temp data
|
result = checkTempData(assemble);
|
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreMsg("质量检验环节出现错误" );
|
assemble.updateById();
|
return result;
|
}
|
//6 temp 2 record
|
|
result = temp2record(assemble);
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreMsg("数据搬运至主数据记录环节出现错误" );
|
assemble.updateById();
|
return result;
|
}
|
|
//7 record 链接 版本
|
result = linkMaintain(assemble);
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreMsg("联接版本出现错误" );
|
assemble.updateById();
|
return result;
|
}
|
|
//todo 添加事务 log记录
|
//8 更新 参数
|
result = updateParams(assemble);
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreMsg("更新变量环节出现错误" );
|
}else {
|
assemble.setPreMsg("汇集成功");
|
assemble.setPreStatus(SysAssembleRunStatus.success);
|
}
|
assemble.updateById();
|
return result;
|
}
|
catch (Exception e) {
|
assemble.setPreMsg("汇集任务出现错误");
|
assemble.setPreStatus(SysAssembleRunStatus.fail).updateById();
|
e.printStackTrace();
|
|
return Result.error(new CodeMsg(6009, e.getMessage()));
|
}
|
}
|
|
private Result linkMaintain(SysAssemble assemble) {
|
try{
|
String menuId = assemble.getMenuId();
|
MenuMapping menuMapping = menuMappingService.selectOne(new EntityWrapper<MenuMapping>().eq("menu_id", menuId));
|
String tableName = menuMapping.getTableName();
|
Maintain maintain = masterDataService.uploadedData(tableName, assemble.getUpdateType(), assemble.getUserId());
|
|
Result result = temp2recordUpdate(assemble, maintain);
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreMsg("搬运更新字段出现错误" );
|
assemble.updateById();
|
return result;
|
}
|
Boolean audit = menuMapping.getAudit();
|
if (audit == null){
|
//默认需要审批
|
audit = true;
|
}
|
// 处理关联人
|
|
|
if (audit) {
|
String chargeId = menuMapping.getChargeId();
|
TUser user = userService.selectById(chargeId);
|
if (user == null) {
|
return Result.error(new CodeMsg(6009, "找不到对应的负责人:" + chargeId));
|
}
|
activitiService.setUser(user);
|
Flows flows = activitiService.start("process", null, maintain.getId(), ActivitiBusinessType.maintain);
|
assemble.setFlowId(flows.getId()).updateById();
|
maintain.setFlowId(flows.getId());
|
maintain.setDesp("启动汇集流程");
|
maintain.updateById();
|
masterModifiedService.dealAssemble(maintain.getId(), assemble.getUserId(), true);
|
}else {
|
//直接运行
|
Flows flows = new Flows().setStatus(ActivitiStatus.open).setBusinessId(maintain.getId()).setId(DbUtils.getUUID()).setCreateTime(new Date());
|
flows.insert();
|
maintain.setFlowId(flows.getId());
|
maintain.updateById();
|
maintainService.dealFlow(maintain.getId(), ActivitiStatus.open);
|
masterModifiedService.dealAssemble(maintain.getId(), assemble.getUserId(), false);
|
}
|
return Result.success(null);
|
}catch (Exception e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6009, e.getMessage()));
|
}
|
|
}
|
|
private Result updateParams(SysAssemble assemble) {
|
Connection conn = null;
|
try{
|
String id = assemble.getId();
|
Boolean bigdata = assemble.getBigdata();
|
|
DataSourceInfo dataSourceInfo;
|
if (bigdata) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
}
|
conn = dataSourceInfo.conn();
|
List<SysAssembleParams> paramsList = paramsService.selectList(new EntityWrapper<SysAssembleParams>().eq("parent_id", id));
|
|
for (SysAssembleParams sysAssembleParams : paramsList) {
|
String updateSql = sysAssembleParams.getUpdateSql();
|
if (StringUtils.isEmpty(updateSql)){
|
continue;
|
}
|
|
PreparedStatement preparedStatement = conn.prepareStatement(updateSql, ResultSet.TYPE_SCROLL_INSENSITIVE,ResultSet.CONCUR_READ_ONLY);
|
ResultSet resultSet = preparedStatement.executeQuery();
|
//默认只有一个结果
|
if (resultSet.first()) {
|
String val = resultSet.getString(1);
|
sysAssembleParams.setVal(val);
|
sysAssembleParams.updateById();
|
}
|
|
}
|
return Result.success(null) ;
|
}
|
catch (Exception e) {
|
|
return Result.error(new CodeMsg(6006, e.getMessage())) ;
|
}
|
finally {
|
if (conn != null) {
|
try {
|
conn.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6006, e.getMessage())) ;
|
}
|
}
|
}
|
|
}
|
|
private Result checkTempTable(SysAssemble assemble) {
|
try {
|
String menuId = assemble.getMenuId();
|
MenuMapping menuMapping = menuMappingService.selectOne(new EntityWrapper<MenuMapping>().eq("menu_id", menuId));
|
String tableName = menuMapping.getTableName();
|
String tempTableName = Constant.Temp + tableName;
|
Boolean bigdata = assemble.getBigdata();
|
DataSourceInfo dataSourceInfo;
|
if (bigdata) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
}
|
boolean exists = dataSourceInfo.checkTableExists(tempTableName);
|
if (!exists){
|
List<String> fieldsFromTable = dataSourceInfo.getFieldsFromTable(tableName);
|
if (fieldsFromTable == null) {
|
return Result.error(new CodeMsg(6003, "查询正式表的字段:" + tableName));
|
}
|
dataSourceInfo.createTable(tempTableName, fieldsFromTable);
|
}
|
//drop temp data
|
dataSourceInfo.truncateData(tempTableName);
|
return Result.success(null);
|
}catch (Exception e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6004, e.getMessage()));
|
}
|
}
|
private Result temp2recordUpdate(SysAssemble assemble, Maintain maintain) {
|
Connection conn = null;
|
try {
|
String menuId = assemble.getMenuId();
|
Boolean bigdata = assemble.getBigdata();
|
MenuMapping menuMapping = menuMappingService.selectOne(new EntityWrapper<MenuMapping>().eq("menu_id", menuId));
|
String tableName = menuMapping.getTableName();
|
String tempTableName = Constant.Temp + tableName;
|
String recordTableName = tableName + Constant.RECORD;
|
DataSourceInfo dataSourceInfo;
|
String updateSql = null;
|
if (bigdata) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
updateSql = Constant.MYSQLJoinUpdateSql;
|
}
|
conn = dataSourceInfo.conn();
|
SysAssembleUpdateType updateType = assemble.getUpdateType();
|
List<String> fieldsFromTable = dataSourceInfo.getFieldsFromTable(tableName);
|
if (fieldsFromTable.contains(Constant.ID)) {
|
fieldsFromTable.remove(Constant.ID);
|
}
|
|
List<String> fieldFromRecordTable = new ArrayList<>(fieldsFromTable);
|
fieldFromRecordTable.add(Constant.ID);
|
fieldFromRecordTable.add(Constant.STD_ID);
|
fieldFromRecordTable.add(Constant.DEAL);
|
|
String recordFieldStr = StringUtils.join(fieldFromRecordTable, Constant.COMMA);
|
int cnt = 0;
|
if(updateType.equals(SysAssembleUpdateType.Increment)) {
|
String updateFields = assemble.getUpdateFields();
|
String[] split = updateFields.split(Constant.SEMICOLON);
|
ArrayList<String> unionCodeFields = new ArrayList<String>(Arrays.asList(split));
|
String joinStr = getJoinFieldParse(unionCodeFields);
|
List<String> otherFields = fieldsFromTable.stream().filter(s -> !unionCodeFields.contains(s)).filter(s -> !s.equalsIgnoreCase(Constant.ID)).collect(Collectors.toList());
|
if (otherFields.size() != 0) {
|
String insertFieldStr = fieldsFromTable.stream()
|
.map(s -> MessageFormat.format(Constant.Alias,Constant.T1,s))
|
.collect(Collectors.joining(Constant.COMMA));
|
|
String updatedSql = MessageFormat.format(updateSql, recordTableName, recordFieldStr,insertFieldStr, tempTableName, tableName, joinStr);
|
PreparedStatement updatedPreparedStatement = conn.prepareStatement(updatedSql);
|
int updateCnt = updatedPreparedStatement.executeUpdate();
|
cnt = updateCnt;
|
tableInfoMapper.insertMatintainDetailFromTemp(DbUtils.quotedStr(maintain.getId()), maintain.getTableName() + Constant.RECORD , DbUtils.quotedStr(Operate.update.toString()));
|
}
|
}
|
|
//tableInfoMapper.updateStdId( maintain.getTableName() + Constant.RECORD);
|
|
tableInfoMapper.tempDeal(maintain.getTableName() + Constant.RECORD, DbUtils.quotedStr(maintain.getId()));
|
|
assemble.setPreCnt(assemble.getPreCnt() + cnt);
|
assemble.updateById();
|
return Result.success(null);
|
}
|
catch (Exception e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6005, e.getMessage()));
|
}
|
finally {
|
if (conn != null) {
|
try {
|
conn.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6005, e.getMessage()));
|
}
|
}
|
}
|
}
|
|
private Result temp2record(SysAssemble assemble) {
|
Connection conn = null;
|
try {
|
String menuId = assemble.getMenuId();
|
Boolean bigdata = assemble.getBigdata();
|
MenuMapping menuMapping = menuMappingService.selectOne(new EntityWrapper<MenuMapping>().eq("menu_id", menuId));
|
String tableName = menuMapping.getTableName();
|
String tempTableName = Constant.Temp + tableName;
|
String recordTableName = tableName + Constant.RECORD;
|
DataSourceInfo dataSourceInfo;
|
String transSql;
|
String updateSql = null;
|
String insertSql = null;
|
if (bigdata) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
transSql = Constant.Temp2RecordHbaseTemplate;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
transSql = Constant.Temp2RecordMySQLTemplate;
|
updateSql = Constant.MYSQLJoinUpdateSql;
|
insertSql = Constant.MYSQLJoinAddSql;
|
}
|
conn = dataSourceInfo.conn();
|
SysAssembleUpdateType updateType = assemble.getUpdateType();
|
List<String> fieldsFromTable = dataSourceInfo.getFieldsFromTable(tableName);
|
if (fieldsFromTable.contains(Constant.ID)) {
|
fieldsFromTable.remove(Constant.ID);
|
}
|
List<String> fieldFromRecordTable = new ArrayList<>(fieldsFromTable);
|
fieldFromRecordTable.add(Constant.ID);
|
fieldFromRecordTable.add(Constant.STD_ID);
|
fieldFromRecordTable.add(Constant.DEAL);
|
|
String recordFieldStr = StringUtils.join(fieldFromRecordTable, Constant.COMMA);
|
String fieldStr = fieldsFromTable.stream()
|
.collect(Collectors.joining(Constant.COMMA));
|
|
int cnt = 0;
|
if (updateType.equals(SysAssembleUpdateType.All)) {
|
String sql = MessageFormat.format(transSql, recordTableName, recordFieldStr, fieldStr, tempTableName);
|
PreparedStatement preparedStatement = conn.prepareStatement(sql);
|
cnt = preparedStatement.executeUpdate();
|
}else if(updateType.equals(SysAssembleUpdateType.Increment)) {
|
String updateFields = assemble.getUpdateFields();
|
String[] split = updateFields.split(Constant.SEMICOLON);
|
ArrayList<String> unionCodeFields = new ArrayList<String>(Arrays.asList(split));
|
String joinStr = getJoinFieldParse(unionCodeFields);
|
|
// insert
|
String insertFieldStr = fieldsFromTable.stream()
|
.map(s -> MessageFormat.format(Constant.Alias,Constant.T1,s))
|
.collect(Collectors.joining(Constant.COMMA));
|
|
String insetSql = MessageFormat.format(insertSql, recordTableName, recordFieldStr,insertFieldStr, tempTableName, tableName, joinStr);
|
PreparedStatement preparedStatement = conn.prepareStatement(insetSql);
|
int insertCnt = preparedStatement.executeUpdate();
|
cnt += insertCnt;
|
}
|
assemble.setPreCnt(cnt);
|
assemble.updateById();
|
return Result.success(null);
|
}
|
catch (Exception e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6005, e.getMessage()));
|
}
|
finally {
|
if (conn != null) {
|
try {
|
conn.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6005, e.getMessage()));
|
}
|
}
|
}
|
}
|
|
private String getOthetFieldParse(List<String> unionCodeFields) {
|
ContentBuilder builder = new ContentBuilder(Constant.AND);
|
for (String s : unionCodeFields) {
|
builder.append(MessageFormat.format(Constant.MYSQLJoinParse, s));
|
}
|
return builder.toString();
|
}
|
|
private String getJoinFieldParse(ArrayList<String> updateFields) {
|
|
ContentBuilder builder = new ContentBuilder(Constant.AND);
|
for (String s : updateFields) {
|
builder.append(MessageFormat.format(Constant.MYSQLJoinParse, s));
|
}
|
return builder.toString();
|
}
|
|
private Result checkTempData(SysAssemble assemble) {
|
//TODO
|
|
SysAssembleCheckType checkType = assemble.getCheckType();
|
if (checkType == null) {
|
return Result.error(new CodeMsg(6009,"规则校验类型为空"));
|
}
|
if (checkType.equals(SysAssembleCheckType.unlimited)) {
|
return Result.success(null);
|
}
|
|
String menuId = assemble.getMenuId();
|
MenuMapping menuMapping = menuMappingService.selectOne(new EntityWrapper<MenuMapping>().eq("menu_id", menuId));
|
String tableName = menuMapping.getTableName();
|
String tempTableName = Constant.Temp + tableName;
|
HashMap<String,Boolean> fieldResultSet = ruleClient.execuImmeForCollect(tempTableName, assemble.getUserId());
|
switch (checkType){
|
case successAdd:
|
String unSuccessFields = fieldResultSet.keySet().stream().filter(s -> !fieldResultSet.get(s)).collect(Collectors.joining(Constant.COMMA));
|
if (fieldResultSet.keySet().contains(false)) {
|
return Result.error(new CodeMsg(6009,"规则校验不通过 字段:" + unSuccessFields));
|
}
|
break;
|
case partSuccessAdd:
|
String checkFields = assemble.getCheckFields();
|
String[] split = checkFields.split(Constant.SEMICOLON);
|
for (String s : split) {
|
Boolean checked = fieldResultSet.get(s);
|
if (checked == null || !checked) {
|
return Result.error(new CodeMsg(6009,"规则校验不通过 字段:" + s));
|
}
|
}
|
break;
|
}
|
|
|
|
return Result.success(null);
|
}
|
|
private Result purgeData(String purgeSql, String assembleId, boolean bigData) {
|
Connection connection = null;
|
try {
|
DataSourceInfo dataSourceInfo = null;
|
if (bigData) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
}
|
connection = dataSourceInfo.conn();
|
|
List<SysAssembleParams> paramsList = paramsService.selectList(new EntityWrapper<SysAssembleParams>().eq("parent_id", assembleId));
|
HashMap<String, String> paramsMap = new HashMap<>();
|
Set<String> matcher = DbUtils.matcher(purgeSql);
|
for (String s : matcher) {
|
List<SysAssembleParams> collect = paramsList.stream().filter(sysAssembleParams -> sysAssembleParams.getCode().equalsIgnoreCase(s)).collect(Collectors.toList());
|
if (collect.size() == 0) {
|
return Result.error(new CodeMsg(6006, "未匹配到参数:" + s));
|
}else if (collect.size() > 1) {
|
return Result.error(new CodeMsg(6006, "匹配到多个参数:" + s));
|
}else {
|
SysAssembleParams sysAssembleParams = collect.get(0);
|
String val = sysAssembleParams.getVal();
|
if (StringUtils.isEmpty(val)) {
|
String initSql = sysAssembleParams.getInitSql();
|
if (StringUtils.isEmpty(initSql)) {
|
return Result.error(new CodeMsg(6006, "初始化参数:" + s + "失败"));
|
}
|
PreparedStatement preparedStatement = connection.prepareStatement(initSql);
|
ResultSet resultSet = preparedStatement.executeQuery();
|
//默认只有一个结果
|
val = resultSet.getString(1);
|
}
|
paramsMap.put(s, val);
|
}
|
}
|
for (String key : paramsMap.keySet()) {
|
String val = paramsMap.get(key);
|
String preParams = MessageFormat.format(Constant.ParamsShell, key);
|
purgeSql = purgeSql.replace(preParams, val);
|
}
|
|
PreparedStatement preparedStatement = connection.prepareStatement(purgeSql);
|
preparedStatement.execute();
|
return Result.success(null);
|
|
} catch (SQLException e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6004, "运行清洗sql出错," + e.getSQLState()));
|
} finally {
|
if (connection != null) {
|
try {
|
connection.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6003, "本地connection 无法关闭"));
|
}
|
}
|
}
|
}
|
|
private Result runByApi(String id) {
|
//TODO
|
return Result.success(null);
|
}
|
|
private Result runByDb(SysAssemble assemble) {
|
String id = assemble.getId();
|
Boolean bigdata = assemble.getBigdata();
|
|
List<SysAssembleDb> dbList = dbService.selectList(new EntityWrapper<SysAssembleDb>().eq(Constant.PARENT_ID, id).eq(Constant.Active, true));
|
for (SysAssembleDb sysAssembleDb : dbList) {
|
String dbId = sysAssembleDb.getId();
|
Result result = loadOneDb(sysAssembleDb, bigdata, id);
|
if (!result.getSuccess()) {
|
assemble.setPreMsg("导入源数据错误:" + sysAssembleDb.getDatabaseName()).updateById();
|
return result;
|
}
|
}
|
return Result.success(null);
|
}
|
|
private Result loadOneDb(SysAssembleDb sysAssembleDb, boolean bigData, String assembleId) {
|
String dbId = sysAssembleDb.getId();
|
Connection conn = dbService.getConnection(dbId);
|
if (conn == null) {
|
return Result.error(new CodeMsg(6002, MessageFormat.format("未能连接到源 id: {0}", dbId)));
|
}
|
try {
|
List<SysAssembleDbTable> tableList = tableService.selectList(new EntityWrapper<SysAssembleDbTable>().eq(Constant.PARENT_ID, dbId).eq(Constant.Active, true));
|
for (SysAssembleDbTable dbTable : tableList) {
|
String tableId = dbTable.getId();
|
String tempTableName = dbTable.getTempTableName();
|
|
List<SysAssembleDbField> fieldList = fieldService.selectList(new EntityWrapper<SysAssembleDbField>().eq(Constant.PARENT_ID, tableId));
|
String fields = fieldList.stream().map(SysAssembleDbField::getField).collect(Collectors.joining(Constant.COMMA));
|
SysAssembleTableType type = dbTable.getType();
|
String tableName = null;
|
String assembleTempTableName = tempTableName;
|
if (type.equals(SysAssembleTableType.sql)) {
|
tableName = dbTable.getSql();
|
} else if (type.equals(SysAssembleTableType.table)) {
|
tableName = dbTable.getTableName();
|
}
|
//TODO assembleTempTableName 可能会超长 64字节 后续修改
|
DataSourceInfo dataSourceInfo = null;
|
if (!bigData) {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}
|
|
// drop temp data
|
dataSourceInfo.truncateData(tempTableName);
|
|
fixAssembleTempTable(dataSourceInfo, assembleTempTableName, fieldList, dbId);
|
|
//checkAssembleTempTableExists(assembleTempTableName);
|
//TODO 未分页
|
String filter = dbTable.getFilter();
|
Set<String> matcher = DbUtils.matcher(filter);
|
for (String code : matcher) {
|
SysAssembleParams sysAssembleParams = paramsService.selectOne(new EntityWrapper<SysAssembleParams>().eq(Constant.PARENT_ID, assembleId).eq(Constant.Code, code));
|
if (sysAssembleParams == null){
|
return Result.error(new CodeMsg(6009, assembleId + "有变量未匹配到:"+ code));
|
}
|
String val = sysAssembleParams.getVal();
|
if (StringUtils.isEmpty(val)) {
|
return Result.error(new CodeMsg(6009, assembleId + "有变量未获取到值:"+ code));
|
}
|
val = DbUtils.quotedStr(val);
|
filter = filter.replace(DbUtils.assemblParam(code), val);
|
}
|
if (StringUtils.isEmpty(filter)) {
|
filter = Constant.WHERE_DEFAULT;
|
}
|
String runSqlTemplate = null;
|
if (type.equals(SysAssembleTableType.table)){
|
runSqlTemplate = Constant.selectFieldTableTemplate;
|
}else if(type.equals(SysAssembleTableType.sql)){
|
runSqlTemplate = Constant.selectFieldSqlTemplate;
|
}
|
String sql = MessageFormat.format(runSqlTemplate, fields, tableName, filter);
|
PreparedStatement ps = conn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
|
ps.setFetchSize(Integer.MIN_VALUE);
|
ps.setFetchDirection(ResultSet.FETCH_REVERSE);
|
ResultSet resultSet = ps.executeQuery();
|
ResultSetMetaData metaData = resultSet.getMetaData();
|
int columnCount = metaData.getColumnCount();
|
|
String insertMySqlSql = assembleSql(unBigDataDataSourceInfo.getDbName(), assembleTempTableName, fields);
|
int cnt = 0;
|
StringBuilder builder = new StringBuilder();
|
while (resultSet.next()) {
|
|
for (int i = 1; i <= columnCount; i++) {
|
if (i == columnCount) {
|
builderEnd(builder, resultSet.getObject(i));
|
}else {
|
builderAppend(builder, resultSet.getObject(i));
|
}
|
|
}
|
cnt++;
|
if (cnt == 5000) {
|
runOneBatch(bigData, insertMySqlSql, builder);
|
}
|
}
|
if (builder.length() > 0) {
|
runOneBatch(bigData, insertMySqlSql, builder);
|
}
|
}
|
return Result.success(null);
|
}
|
catch (Exception e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6004, e.getMessage()));
|
}finally {
|
if (conn != null) {
|
try {
|
conn.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
}
|
}
|
}
|
|
}
|
|
private void runOneBatch(boolean bigData, String insertMySqlSql, StringBuilder builder) throws UnsupportedEncodingException, SQLException {
|
byte[] bytes = builder.toString().getBytes("UTF-8");
|
if (bytes.length > 0) {
|
InputStream stream = new ByteArrayInputStream(bytes);
|
bulkLoadFromInputStream(insertMySqlSql, stream, bigData);
|
}
|
//清空bulider
|
builder.delete(0, builder.length());
|
}
|
|
private void fixAssembleTempTable(DataSourceInfo dataSourceInfo, String assembleTempTableName, List<SysAssembleDbField> dbFieldList, String dbId) {
|
boolean exists = dataSourceInfo.checkTableExists(assembleTempTableName);
|
List<String> fieldList = dbFieldList.stream().map(SysAssembleDbField::getField).collect(Collectors.toList());
|
if (!exists) {
|
dataSourceInfo.createTable(assembleTempTableName, fieldList);
|
}else {
|
//check 字段是否相同
|
boolean checked = dataSourceInfo.checkFieldList(assembleTempTableName, fieldList);
|
if (!checked){
|
assembleTempTableName = reSetTableName(assembleTempTableName, dbId);
|
dataSourceInfo.createTable(assembleTempTableName, fieldList);
|
|
}
|
}
|
}
|
|
private String reSetTableName(String assembleTempTableName, String tableId) {
|
//TODO Db
|
SysAssembleDbTable dbTable = tableService.selectById(tableId);
|
String suffix = null;
|
suffix = DbUtils.getUUID(16);
|
String tempTableName = null;
|
if (dbTable.getType().equals(SysAssembleTableType.table)) {
|
tempTableName = Constant.AssembleTempTable + dbTable.getTableName() + suffix;
|
}else {
|
tempTableName = Constant.AssembleTempSql+ DbUtils.getUUID(5)+ suffix;
|
}
|
|
dbTable.setTempTableName(tempTableName).updateById();
|
return tempTableName;
|
}
|
|
public void builderEnd(StringBuilder builder, Object object) {
|
builder.append(object);
|
builder.append("\n");
|
}
|
|
|
public void builderAppend(StringBuilder builder, Object object) {
|
builder.append(object);
|
builder.append("\t");
|
}
|
|
|
public String assembleSql(String dataBaseName, String tableName, String fields) {
|
String sql = "LOAD DATA LOCAL INFILE 'sql.csv' IGNORE INTO TABLE " + dataBaseName + "." + tableName + "(" + fields + ")";
|
return sql;
|
}
|
|
private int bulkLoadFromInputStream(String sql, InputStream dataStream, boolean bigData) {
|
if (null == dataStream) {
|
log.error("输入流为NULL，没有数据导入。");
|
return 0;
|
}
|
Connection conn = null;
|
try {
|
DataSourceInfo dataSourceInfo = null;
|
if (bigData) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
}
|
|
conn = dataSourceInfo.conn();
|
PreparedStatement statement = conn.prepareStatement(sql);
|
int result = 0;
|
if (statement.isWrapperFor(com.mysql.jdbc.Statement.class)) {
|
com.mysql.jdbc.PreparedStatement mysqlStatement = statement.unwrap(com.mysql.jdbc.PreparedStatement.class);
|
mysqlStatement.setLocalInfileInputStream(dataStream);
|
result = mysqlStatement.executeUpdate();
|
}
|
return result;
|
}
|
catch (Exception e) {
|
e.printStackTrace();
|
}
|
finally {
|
if (conn != null) {
|
try {
|
conn.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
}
|
}
|
}
|
return 0;
|
}
|
}
|
