Skip to content

J
jz-dmp-service
Commit f353dd2b authored by mcb's avatar mcb
Browse files
Options

no message

parent 61ebabf8
Show whitespace changes
Inline Side-by-side
Showing with 292 additions and 153 deletions
src/main/java/com/jz/dmp/modules/dao/DmpDevelopTaskDao.java
View file @ f353dd2b
......@@ -15,7 +15,7 @@ public interface DmpDevelopTaskDao {
int deleteNavigationTreeByTreeId(String treeId) throws Exception;
Integer getDbInfoByParam(@Param("xmlTdb") String xmlTdb, @Param("projectId") Integer projectId);
Integer getDbInfoByParam(@Param("xmlTdb") String xmlTdb, @Param("projectId") Integer projectId) throws Exception;
int insert(DmpDevelopTask dmpDevelopTask) throws Exception;
......
src/main/java/com/jz/dmp/modules/service/impl/DmpDevelopTaskServiceImpl.java
View file @ f353dd2b
......@@ -67,7 +67,7 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
/**
* 提交离线同步到指定projectId项目下（上传xml配置文件）
* */
*/
private JsonResult submitSyncXml(long projectId, String xmlContent) {
String _type_ = XmlUtils.getPropertyValue(xmlContent, "type");
String _fileName_ = XmlUtils.getPropertyValue(xmlContent, "name");
......@@ -106,6 +106,7 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
return new JsonResult(ResultCode.SUCCESS, rst);
}
}
/*
* 生成xml
* */
......@@ -131,10 +132,10 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
String _executorMemory = (String) settingMap.get("executorMemory");
String _executorCores = (String) settingMap.get("executorCores");
String _totalExecutorCores = (String) settingMap.get("totalExecutorCores");
String _ftColumn =(String) settingMap.get("ftColumn");//分桶字段
String _ftCount =(String) settingMap.get("ftCount");//分桶个数
String _separateMax =(String) settingMap.get("separateMax");//分桶字段最大值
String _separateMin =(String) settingMap.get("separateMin");//分桶字段最小值
String _ftColumn = (String) settingMap.get("ftColumn");//分桶字段
String _ftCount = (String) settingMap.get("ftCount");//分桶个数
String _separateMax = (String) settingMap.get("separateMax");//分桶字段最大值
String _separateMin = (String) settingMap.get("separateMin");//分桶字段最小值
//******源数据******
Map<String, Object> readerMap = (Map<String, Object>) scriptMap.get("reader");
......@@ -168,10 +169,10 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
List<Map<String, Object>> _writerColumns = (List<Map<String, Object>>) writerMap.get("column");//目标数据库表字段
Integer _projectId_ = body.getProjectId(); //项目id
Integer _parentId_ = (Integer) map.get("parentId"); //父节点id
/*Integer _parentId_ = (Integer) map.get("parentId"); //父节点id
String _mode_ = (String) map.get("mode");
String _version_ = (String) map.get("version");
String _name_ = (String) map.get("name"); //节点数名称
String _name_ = (String) map.get("name"); //节点数名称*/
//根据项目id和数据源名称 查询数据信息
DmpSyncingDatasource sDS = this.getDmpSyncingDatasource(_projectId_, _dbConnection);//源
......@@ -194,7 +195,7 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
boolean isBigData2ftp = ("HIVE".equals(st) || "KUDU".equals(st)) && "FTP".equals(tt);
if ("RDBMS".equals(sDST.getDatasourceCatecode()) && "HIVE".equals(tt)) {
_registerTableName = sDS.getDatasourceName() + "_" + _registerTableName.substring(_registerTableName.indexOf(".")+1);
_registerTableName = sDS.getDatasourceName() + "_" + _registerTableName.substring(_registerTableName.indexOf(".") + 1);
}
String name_ = (StringUtils.hasLength(_targetDbName_) ? _targetDbName_ : tDS.getDatasourceName()) + "_" + _targetTable;
......@@ -222,7 +223,7 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
if ("RDBMS".equals(sDST.getDatasourceCatecode())) {
sb.append("<source_table db_connection=\"").append(sDS.getDatasourceName()).append("\" register_table_name=\"")
.append(_registerTableName.substring(_registerTableName.indexOf(".")+1)).append("\">").append(source_table_).append("</source_table>").append("\r\n");
.append(_registerTableName.substring(_registerTableName.indexOf(".") + 1)).append("\">").append(source_table_).append("</source_table>").append("\r\n");
}
if ("HIVE".equals(st) || "KUDU".equals(st)) {
String sourceTableName = _registerTableName;
......@@ -299,6 +300,7 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
}
return sb.toString();
}
/*
*根据项目id和数据源名称 查询数据信息
* */
......@@ -308,7 +310,7 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
ds.setDatasourceName(dsName);
ds.setProjectId(projectId);
List<DmpSyncingDatasource> dsList = dmpSyncingDatasourceService.findListByParams(ds);
if(dsList != null && dsList.size() > 0) {
if (dsList != null && dsList.size() > 0) {
return dsList.get(0);
} else {
return null;
......@@ -319,17 +321,17 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
* 获取文件类型
* */
private String getFileType(String ft, String _sourceHdfsFile, String _sourceFtpFile) {
if(StringUtils.isBlank(ft)) {
if(StringUtils.isNotBlank(_sourceHdfsFile)) {
if (StringUtils.isBlank(ft)) {
if (StringUtils.isNotBlank(_sourceHdfsFile)) {
int idx = _sourceHdfsFile.lastIndexOf(".");
if(idx > -1) {
String nft = _sourceHdfsFile.substring(idx+1, _sourceHdfsFile.length());
if(StringUtils.isNotBlank(nft)) {
if("csv".equalsIgnoreCase(nft)) {
if (idx > -1) {
String nft = _sourceHdfsFile.substring(idx + 1, _sourceHdfsFile.length());
if (StringUtils.isNotBlank(nft)) {
if ("csv".equalsIgnoreCase(nft)) {
ft = "csv";
} else if("json".equalsIgnoreCase(nft)) {
} else if ("json".equalsIgnoreCase(nft)) {
ft = "json";
} else if("txt".equalsIgnoreCase(nft)) {
} else if ("txt".equalsIgnoreCase(nft)) {
ft = "csv";
} else {
ft = "csv";
......@@ -337,16 +339,16 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
}
}
}
if(StringUtils.isNotBlank(_sourceFtpFile)) {
if (StringUtils.isNotBlank(_sourceFtpFile)) {
int idx = _sourceFtpFile.lastIndexOf(".");
if(idx > -1) {
String nft = _sourceFtpFile.substring(idx+1, _sourceFtpFile.length());
if(StringUtils.isNotBlank(nft)) {
if("csv".equalsIgnoreCase(nft)) {
if (idx > -1) {
String nft = _sourceFtpFile.substring(idx + 1, _sourceFtpFile.length());
if (StringUtils.isNotBlank(nft)) {
if ("csv".equalsIgnoreCase(nft)) {
ft = "csv";
} else if("json".equalsIgnoreCase(nft)) {
} else if ("json".equalsIgnoreCase(nft)) {
ft = "json";
} else if("txt".equalsIgnoreCase(nft)) {
} else if ("txt".equalsIgnoreCase(nft)) {
ft = "csv";
} else {
ft = "csv";
......@@ -360,21 +362,21 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
private String getTargetDBname(DmpSyncingDatasource ds) {
String url = ds.getJdbcUrl();
if(StringUtils.isNotBlank(url)) {
if(ds.getDatasourceType() ==2 ){//sqlserver
if (StringUtils.isNotBlank(url)) {
if (ds.getDatasourceType() == 2) {//sqlserver
String[] arr1 = url.split("=");
String[] arr2 = arr1[1].split(";");
System.out.println( arr2[0]);
System.out.println(arr2[0]);
return arr2[0];
}else{
} else {
String[] arr1 = url.split("/");
String ls = arr1[arr1.length-1];
String ls = arr1[arr1.length - 1];
String[] arr2 = ls.split("\\?");
return arr2[0];
}
} else {
if(ds.getDatasourceType().equals("11")){//ftp
if (ds.getDatasourceType().equals("11")) {//ftp
return ds.getDbName();
}
return "";
......@@ -385,10 +387,10 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
String sT = s.getDatasourceType().toLowerCase();
String tT = t.getDatasourceType().toLowerCase();
String _ft = "";
if(StringUtils.isNotBlank(ft)) {
if (StringUtils.isNotBlank(ft)) {
_ft = ft;
}
if(s.getDatasourceCatecode().equals("RDBMS")) {
if (s.getDatasourceCatecode().equals("RDBMS")) {
sT = "jdbc";
}
return sT + _ft + "2" + tT;
......@@ -425,13 +427,13 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
String name = (String) m.get("name");
String type = StringUtils.getStringValue(m.get("type"), "");//relatedTypeMap.get(StringUtils.getStringValue(m.get("type"), ""));
String isPt = (String) m.get("isPt");
if("SQLSERVER".equals(s.getDatasourceType())){ //对sqlserver 修改
if("STRING".equals(type)){
source_table += "REPLACE(REPLACE(REPLACE("+name+", CHAR(13), '<br1>'),CHAR(10),'<br2>'),',','<comma>') AS "+name;
}else{
if ("SQLSERVER".equals(s.getDatasourceType())) { //对sqlserver 修改
if ("STRING".equals(type)) {
source_table += "REPLACE(REPLACE(REPLACE(" + name + ", CHAR(13), '<br1>'),CHAR(10),'<br2>'),',','<comma>') AS " + name;
} else {
source_table += name;
}
}else{
} else {
// source_table += "replace(cast(" + name + " as char), char(10), '') as " + name;
source_table += name;
}
......@@ -445,11 +447,11 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
}
} else {
source_columns += name;
if("varchar".equalsIgnoreCase(type)) {
if ("varchar".equalsIgnoreCase(type)) {
source_sql += "cast(" + name + " as string) as ";
} else if("char".equalsIgnoreCase(type)) {
} else if ("char".equalsIgnoreCase(type)) {
source_sql += "cast(" + name + " as string) as ";
} else if("datetime".equalsIgnoreCase(type)) {
} else if ("datetime".equalsIgnoreCase(type)) {
source_sql += "cast(" + name + " as timestamp) as ";
} else {
source_sql += "cast(" + name + " as " + type + ") as ";
......@@ -524,13 +526,13 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
}
private String getSourceHdfsDir(String filePath, String fileName) {
if(StringUtils.isNotBlank(fileName)) {
if (StringUtils.isNotBlank(fileName)) {
boolean f1 = filePath.endsWith("/");
boolean f2 = fileName.startsWith("/");
if(f1 && f2) {
if (f1 && f2) {
fileName = fileName.substring(1, fileName.length());
}
if(!f1 && !f2) {
if (!f1 && !f2) {
filePath += "/";
}
return filePath + fileName;
......@@ -582,7 +584,7 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
target_partition_columns += name + ", ";
target_partition_columns_hive += name + " " + type + ", ";
}
String[] rls = this.appendRules(m.get("rules"), name, clean_rules, check_rules);
String[] rls = this.appendRules(m.get("rules"), name, clean_rules, check_rules);//便利清洗规则
clean_rules = rls[0];
check_rules = rls[1];
if ("HIVE".equals(t.getDatasourceType()) && "1".equals(isPt)) {
......@@ -668,29 +670,31 @@ public class DmpDevelopTaskServiceImpl implements DmpDevelopTaskService {
private String[] appendRules(Object obj, String columnName, String clean_rules, String check_rules) {
List<Map<String, Object>> rules = (List<Map<String, Object>>) obj;
for(Map<String, Object> m : rules) {
if (rules.size() > 0 && rules != null) {
for (Map<String, Object> m : rules) {
String method = (String) m.get("method"); // 规则方法
String type = (String) m.get("type"); // 1:清洗规则;0:校验规则
List<String> params = (ArrayList<String>) m.get("params"); // 规则参数列表，字符串
String sub_rules = "\t<column_name name=\""+columnName+"\">\r\n";
sub_rules += "\t\t<rules_type method=\""+method+"\">\r\n";
if(params != null && params.size() > 0)
for(String rv : params) {
String sub_rules = "\t<column_name name=\"" + columnName + "\">\r\n";
sub_rules += "\t\t<rules_type method=\"" + method + "\">\r\n";
if (params != null && params.size() > 0)
for (String rv : params) {
sub_rules += "\t\t\t<rules_value>" + rv + "</rules_value>\r\n";
}
sub_rules += "\t\t</rules_type>\r\n";
sub_rules += "\t</column_name>\r\n";
if("0".equals(type)) {
if(StringUtils.isBlank(check_rules))
if ("0".equals(type)) {
if (StringUtils.isBlank(check_rules))
sub_rules = "\r\n" + sub_rules;
check_rules += sub_rules;
}
if("1".equals(type)) {
if(StringUtils.isBlank(clean_rules))
if ("1".equals(type)) {
if (StringUtils.isBlank(clean_rules))
sub_rules = "\r\n" + sub_rules;
clean_rules += sub_rules;
}
}
return new String[] {clean_rules, check_rules};
}
return new String[]{clean_rules, check_rules};
}
}
\ No newline at end of file
src/main/java/com/jz/dmp/modules/service/impl/OfflineSynchServiceImpl.java
View file @ f353dd2b
......@@ -452,10 +452,10 @@ public class OfflineSynchServiceImpl implements OfflineSynchService {
* @return
*/
public JsonResult addSyncing(Map<String, Object> body) throws Exception {
Integer projectId = (Integer) body.get("projectId");
Integer parentId = (Integer) body.get("parentId"); //父节点ID
Integer projectId = Integer.valueOf(body.get("projectId").toString()) ;
Integer parentId = Integer.valueOf(body.get("parentId").toString()); //父节点ID
String taskName = (String) body.get("taskName"); //任务名称 业务节点名称 一对一
Integer treeId = (Integer) body.get("treeId"); //树节点ID
Integer treeId = Integer.valueOf(body.get("treeId").toString()); //树节点ID
if (StringUtils.isBlank(taskName)) {
return new JsonResult(ResultCode.PARAMS_ERROR, "任务名称不能为空");
......@@ -502,6 +502,7 @@ public class OfflineSynchServiceImpl implements OfflineSynchService {
task.setTaskDesc("Syncing Task"); //任务描述
task.setIsSubmit("0"); //是否已提交
task.setTreeId(treeId);
task.setDataStatus("1");
String script = JsonMapper.toJsonString(body);
byte[] data = null;
try {
......@@ -561,7 +562,7 @@ public class OfflineSynchServiceImpl implements OfflineSynchService {
private void settRuleInfo(String taskId, List<Map> rules, List<DvTaskRuleT> list) throws Exception {
for (Map rule : rules) {
DvTaskRuleT taskRuleT = new DvTaskRuleT();
Integer ruleId = (Integer) rule.get("ruleId");
Integer ruleId = Integer.valueOf(rule.get("ruleId").toString());
String ruleValue = (String) rule.get("ruleValue");
taskRuleT.setTaskId(taskId);//任务ID
taskRuleT.setRuleId(ruleId.longValue());
......
src/main/resources/lxTaskJson.json
View file @ f353dd2b
//{
// "version": "1.0",
// "parentId": "",
// "mode": "0",
// "projectId": "",
// "name": "",
// "scripts": {
// "setting": {
// "extract": "incremental", //增量或全量
// "extractExpression": "where 1=1", //增量表达式
// "targetBucketCounts": "",
// "errorLimitRecord": "0",
// "executorMemory": "",
// "executorCores": "",
// "totalExecutorCores": ""
// },
// "reader": {
// "dbConnection": "",
// "fileType": "",
// "sourceHdfsPath": "",
// "sourceHdfsFile": "",
// "sourceFtpDir": "",
// "sourceFtpFile": "",
// "sourceSkipFtpFile": "",
// "sourceCsvDelimiter": "",
// "sourceCsvHeader": "",
// "sourceCsvCharset": "",
// "sourceCsvQuote": "",
// "sourceFtpLoadDate": "",
// "registerTableName": "",
// "dayByDay": "false",
// "column": [
// {
// "name": "",
// "type": ""
// }
// ]
// },
// "writer": {
// "targetDbConnection": "",
// "targetTable": "",
// "targetFtpDir": "",
// "targetFtpFile": "",
// "targetCsvDelimiter": "",
// "targetCsvHeader": "",
// "targetCsvCharset": "",
// "targetInsertMergeOverwrite": "",
// "column": [
// {
// "name": "",
// "type": "",
// "isPk": "",
// "isPt": "",
// "rules": []
// }
// ]
// }
// },
// "treeId": 0,
// "taskRules": [
// {
// "ruleId": "",
// "ruleValue": {
// "dv_fields": [
// {
// "fieldName": ""
// }
// ],
// "dvTime": {
// "timeField": "",
// "timeValue": {
// "startTime": "",
// "endTime": ""
// }
// }
// }
// }
// ]
//}
{
"version": "1.0",
"parentId": "",
"mode": "0",
"projectId": "",
"name": "",
"scripts": {
"setting": {
"extract": "incremental", //增量或全量
"extractExpression": "where 1=1", //增量表达式
"targetBucketCounts": "",
"errorLimitRecord": "0",
"executorMemory": "",
"executorCores": "",
"totalExecutorCores": ""
"params":{
"version":"1.0",
"parentId":"509",
"mode":"0",
"projectId":"31",
"taskName":"dmp_demo_dmp_azkaban_exector_server_config",
"scripts":{
"setting":{
"extract":"incremental",
"extractExpression":"where 1=1",
"targetInsertMergeOverwrite":"insert",
"ftColumn":"",
"ftCount":"20",
"separateMax":"",
"separateMin":"",
"primaryKey":"on",
"partition":"on"
},
"reader": {
"dbConnection": "",
"fileType": "",
"sourceHdfsPath": "",
"sourceHdfsFile": "",
"sourceFtpDir": "",
"sourceFtpFile": "",
"sourceSkipFtpFile": "",
"sourceCsvDelimiter": "",
"sourceCsvHeader": "",
"sourceCsvCharset": "",
"sourceCsvQuote": "",
"sourceFtpLoadDate": "",
"registerTableName": "",
"dayByDay": "false",
"column": [
"reader":{
"dbConnection":"mysql_dmp_demo_test",
"fileType":"",
"sourceHdfsPath":"",
"sourceHdfsFile":"",
"sourceFtpDir":"",
"sourceFtpFile":"",
"sourceSkipFtpFile":"",
"sourceCsvDelimiter":"",
"sourceCsvHeader":"",
"sourceCsvCharset":"",
"sourceCsvQuote":"",
"sourceFtpLoadDate":"",
"registerTableName":"dmp_azkaban_exector_server_config",
"dayByDay":"false",
"column":[
{
"name": "",
"type": ""
"name":"host",
"type":"VARCHAR"
},
{
"name":"port",
"type":"VARCHAR"
},
{
"name":"user_name",
"type":"VARCHAR"
},
{
"name":"pass_word",
"type":"VARCHAR"
}
]
},
"writer": {
"targetDbConnection": "",
"targetTable": "",
"targetFtpDir": "",
"targetFtpFile": "",
"targetCsvDelimiter": "",
"targetCsvHeader": "",
"targetCsvCharset": "",
"targetInsertMergeOverwrite": "",
"column": [
"writer":{
"targetDbConnection":"mysql_dmp_demo",
"targetTable":"dmp_azkaban_exector_server_config",
"targetFtpDir":"",
"targetFtpFile":"",
"targetCsvDelimiter":"",
"targetCsvHeader":"",
"targetCsvCharset":"",
"targetInsertMergeOverwrite":"insert",
"column":[
{
"name": "",
"type": "",
"isPk": "",
"isPt": "",
"rules": []
"name":"host",
"type":"VARCHAR",
"isPk":"1",
"isPt":"0",
"rules":[
{"method":"",
"type":""
}
]
},
{
"name":"port",
"type":"VARCHAR",
"isPk":"0",
"isPt":"1",
"rules":[
{"method":"",
"type":""
}
]
},
"treeId": 0,
"taskRules": [
{
"ruleId": "",
"ruleValue": {
"dv_fields": [
"name":"user_name",
"type":"VARCHAR",
"isPk":"0",
"isPt":"0",
"rules":[
{"method":"",
"type":""
}
]
},
{
"name":"pass_word",
"type":"VARCHAR",
"isPk":"0",
"isPt":"0",
"rules":[
{"method":"",
"type":""
}
]
}
]
}
},
"treeId":669,
"taskRules":[
{
"fieldName": ""
"ruleId":"",
"ruleValue":{
"dv_fields":[
{
"fieldName":""
}
],
"dvTime": {
"timeField": "",
"timeValue": {
"startTime": "",
"endTime": ""
"dvTime":{
"timeField":"",
"timeValue":{
"startTime":"",
"endTime":""
}
}
}
}
]
}
}
\ No newline at end of file
src/main/resources/mapper/dmp/DmpDevelopTaskMapper.xml
View file @ f353dd2b
......@@ -21,7 +21,7 @@
delete from dmp_navigation_tree where id = #{treeId}
</delete>
<select id="getDbType" resultType="java.lang.Integer">
<select id="getDbInfoByParam" resultType="java.lang.Integer">
SELECT
dsd.id AS id
FROM dmp_syncing_datasource dsd
......@@ -108,7 +108,7 @@
where ID = #{id}
</update>
<select id="selectTaskInfoByParam" parameterType="map">
<select id="selectTaskInfoByParam" parameterType="map" resultType="com.jz.dmp.modules.model.DmpDevelopTask">
select
ID, datasource_id, TASK_TYPE, TYPE, SCHEDULE_TYPE, IS_SUBMIT, TASK_DESC, SCRIPT, DATA_STATUS
, CREATE_USER_ID, CREATE_TIME, UPDATE_USER_ID, UPDATE_TIME, TREE_ID, CHK_RESULT, SYNC_RESULT, CHK_TIME, SYNC_TIME, FLOW_HEADER, FLOW_JSON, VERSION, IS_GZIPED
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment