sql api 提交

src/main/java/com/jz/dmp/cmdexectool/common/constant/CommConstant.java

@@ -59,6 +59,7 @@ public class CommConstant {
 	public static final String WATERDROP_FTL_SINK_HDFS = "sink_hdfs.ftl";
 	public static final String WATERDROP_FTL_SINK_JDBC = "sink_jdbc.ftl";
 	public static final String WATERDROP_FTL_SINK_KAFKA = "sink_kafka.ftl";
+	public static final String WATERDROP_FTL_SINK_API = "sink_api.ftl";
 	
 	//其他script模板
 	public static final String FTL_SFTP_DOWNLOAD = "sftp_download.ftl";//ftp下载

src/main/java/com/jz/dmp/cmdexectool/scheduler/common/task/sql/SqlParameters.java

@@ -25,6 +25,7 @@ import com.jz.dmp.cmdexectool.entity.DmpSyncingDatasource;
 import com.jz.dmp.cmdexectool.mapper.DmpSyncingDatasourceDao;
 import com.jz.dmp.cmdexectool.scheduler.common.process.ResourceInfo;
 import com.jz.dmp.cmdexectool.scheduler.common.task.AbstractParameters;
+import com.jz.dmp.cmdexectool.scheduler.common.utils.ParameterUtils;
 import com.jz.dmp.cmdexectool.scheduler.server.entity.TaskExecutionContext;
 
 import java.util.HashMap;
@@ -133,18 +134,17 @@ public class SqlParameters extends AbstractParameters {
 			//transform
 			Map<String, String> transformSqlModel = new HashMap<String, String>();
 			transformSqlModel.put("sql", sqlScript);
+			transformSqlModel.put("table_name", "t_view");
 			transform = FreeMarkerUtils.freemakerJson(CommConstant.WATERDROP_FTL_TRANSFORM_SQL, transformSqlModel, freeMarkerConfig);
 			
 			JSONObject tableObj = scriptObj.getJSONObject("table");
-			JSONArray fieldsArr = tableObj.getJSONArray("tableFields");
-			for (int index = 0; index < fieldsArr.size(); index++) {
-				JSONObject fieldObj = fieldsArr.getJSONObject(index);
-				
-				Map<String, String> transformJson2Model = new HashMap<String, String>();
-				transformJson2Model.put("source_field", fieldObj.getString("sourceFieldName"));
-				transformJson2Model.put("target_field", fieldObj.getString("targetFieldName"));
-				transform = transform + FreeMarkerUtils.freemakerJson(CommConstant.WATERDROP_FTL_TRANSFORM_JSON2, transformJson2Model, freeMarkerConfig);
-			}
+			String tableFieldsObj = tableObj.getString("tableFields");
+			String sqlStr = ParameterUtils.columnMappingHandler(tableFieldsObj);
+			
+			Map<String, String> transformMappingSqlModel = new HashMap<String, String>();
+			transformSqlModel.put("sql", sqlStr);
+			transform = transform + FreeMarkerUtils.freemakerJson(CommConstant.WATERDROP_FTL_TRANSFORM_JSON2, transformMappingSqlModel, freeMarkerConfig);
+			
 			//sink
 			
 			//targetSource
@@ -175,7 +175,28 @@ public class SqlParameters extends AbstractParameters {
 			kafkaModel.put("broker", topicObj.getString("server"));
 			sink = FreeMarkerUtils.freemakerJson(CommConstant.WATERDROP_FTL_SINK_KAFKA, kafkaModel, freeMarkerConfig);
 		}else if (CommConstant.OUTPUT_TYPE_API.equals(outputType)) {
+			//transform
+			Map<String, String> transformSqlModel = new HashMap<String, String>();
+			transformSqlModel.put("sql", sqlScript);
+			transformSqlModel.put("table_name", "t_view");
+			transform = FreeMarkerUtils.freemakerJson(CommConstant.WATERDROP_FTL_TRANSFORM_SQL, transformSqlModel, freeMarkerConfig);
+			
+			JSONObject apiObj = scriptObj.getJSONObject("api");
+			String columnFieldsObj = apiObj.getString("columnFields");
+			String sqlStr = ParameterUtils.columnMappingHandler(columnFieldsObj);
 			
+			Map<String, String> transformMappingSqlModel = new HashMap<String, String>();
+			transformSqlModel.put("sql", sqlStr);
+			transform = transform + FreeMarkerUtils.freemakerJson(CommConstant.WATERDROP_FTL_TRANSFORM_JSON2, transformMappingSqlModel, freeMarkerConfig);
+			
+			//sink
+			Map<String, String> sinkApiModel = new HashMap<String, String>();
+			sinkApiModel.put("url", apiObj.getString("apiUrl"));
+			sinkApiModel.put("apiKey", apiObj.getString("apiKey"));
+			sinkApiModel.put("method", apiObj.getString("method"));
+			sinkApiModel.put("signType", apiObj.getString("signType"));
+			sinkApiModel.put("authCode", apiObj.getString("authCode"));
+			sink = FreeMarkerUtils.freemakerJson(CommConstant.WATERDROP_FTL_SINK_API, sinkApiModel, freeMarkerConfig);
 		}
 		//waterdrop script
 		Map<String, String> waterdropModel = new HashMap<String, String>();

src/main/java/com/jz/dmp/cmdexectool/scheduler/common/utils/ParameterUtils.java

@@ -18,16 +18,21 @@
 package com.jz.dmp.cmdexectool.scheduler.common.utils;
 
 import java.sql.PreparedStatement;
+import java.util.ArrayList;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
 import com.jz.dmp.cmdexectool.scheduler.common.Constants;
 import com.jz.dmp.cmdexectool.scheduler.common.enums.CommandType;
 import com.jz.dmp.cmdexectool.scheduler.common.enums.DataType;
@@ -36,6 +41,8 @@ import com.jz.dmp.cmdexectool.scheduler.common.utils.placeholder.BusinessTimeUti
 import com.jz.dmp.cmdexectool.scheduler.common.utils.placeholder.PlaceholderUtils;
 import com.jz.dmp.cmdexectool.scheduler.common.utils.placeholder.TimePlaceholderUtils;
 
+import freemarker.core.ReturnInstruction.Return;
+
 /**
  * parameter parse utils
  */
@@ -248,4 +255,56 @@ public class ParameterUtils {
         }
         return map;
     }
+    
+    /**  
+    * @Title: columnMappingHandler  
+    * @Description: TODO(开发任务SQL：字段映射处理)  
+    * @param @param jsonStr
+    * @param @return    参数  
+    * @return List<Map<String,String>>    返回类型  
+    * @throws  
+    */ 
+    public static String columnMappingHandler(String jsonStr) {
+    	JSONObject jsonObject = JSONObject.parseObject(jsonStr);
+    	JSONArray sourceArray = jsonObject.getJSONArray("sourceFields");
+    	Map<String, JSONObject> sourceMap = new HashMap<String, JSONObject>();
+    	for (int index=0; index<sourceArray.size(); index++) {
+			JSONObject sourceObj = sourceArray.getJSONObject(index);
+			sourceMap.put(sourceObj.getString("customSoruceFiledId"), sourceObj);
+		}
+    	JSONArray targetArray = jsonObject.getJSONArray("targetFields");
+    	Map<String, JSONObject> targetMap = new HashMap<String, JSONObject>();
+    	for (int index=0; index<targetArray.size(); index++) {
+			JSONObject targetObj = targetArray.getJSONObject(index);
+			targetMap.put(targetObj.getString("customTargetFieldId"), targetObj);
+		}
+    	
+    	StringBuilder sb = new StringBuilder("SELECT ");
+    	
+    	JSONArray mappingArray = jsonObject.getJSONArray("columnMapping");
+    	Integer size = mappingArray.size();
+    	for (int index=0; index<size; index++) {
+    		JSONObject mappingObj = mappingArray.getJSONObject(index);
+    		JSONObject sourceObj = sourceMap.get(mappingObj.getString("customSoruceFiledId"));
+    		JSONObject targetObj = sourceMap.get(mappingObj.getString("customTargetFiledId"));
+    		
+    		//String customSoruceFiledId = sourceObj.getString("customSoruceFiledId");
+    		String sourceFieldName = sourceObj.getString("sourceFieldName");
+    		//String sourceFieldType = sourceObj.getString("sourceFieldType");
+    		//String customTargetFieldId = targetObj.getString("customTargetFieldId");
+    		String targetFieldName = targetObj.getString("targetFieldName");
+    		//String targetFieldType = targetObj.getString("targetFieldType");
+    		
+    		sb.append(sourceFieldName +" AS "+targetFieldName);
+    		if (index<size-1) {
+				sb.append(", ");
+			}else {
+				sb.append(" ");
+			}
+		}
+    	sb.append("FROM t_view");
+    	
+    	return sb.toString();
+    }
+
 }

src/main/resources/templates/sink_api.ftl

+Api {
+<#if url??>
+    url = "${url!}"
+</#if>
+<#if apiKey??>
+    apiKey = "${apiKey!}"
+</#if>
+<#if method??>
+    method = "${method!}"
+</#if>
+<#if signType??>
+    signType = "${signType!}"
+</#if>
+<#if authCode??>
+    authCode = "${authCode!}"
+</#if>
+    isTest = "false"
+}

src/test/java/com/jz/cmdexectool/test/task/shell/Test.java

@@ -8,16 +8,20 @@ public class Test {
 	
 	public static void main(String[] args) {
 		try {
-			String fileName = "E:/123/test/test_node.bat";
-			Path path = new File(fileName).toPath();
-			//Files.createDirectories(path);
-			//Files.createFile(path);
-			File file = path.toFile();
-			File parentFile = file.getParentFile();
-			if (!parentFile.exists()) {
-				parentFile.mkdirs();
+			/*
+			 * String fileName = "E:/123/test/test_node.bat"; Path path = new
+			 * File(fileName).toPath(); //Files.createDirectories(path);
+			 * //Files.createFile(path); File file = path.toFile(); File parentFile =
+			 * file.getParentFile(); if (!parentFile.exists()) { parentFile.mkdirs(); }
+			 * file.createNewFile();
+			 */
+			
+			String aa = "E:/aa.csv";
+			String[] arrs = aa.split("\\.");
+			for (String str : arrs) {
+				System.out.println(str);
 			}
-			file.createNewFile();
+			
 		} catch (Exception e) {
 			e.printStackTrace();
 		}