diff --git a/datax-executor/src/main/java/com/wugui/datax/executor/service/command/BuildCommand.java b/datax-executor/src/main/java/com/wugui/datax/executor/service/command/BuildCommand.java index 328ca846..2e0f7535 100644 --- a/datax-executor/src/main/java/com/wugui/datax/executor/service/command/BuildCommand.java +++ b/datax-executor/src/main/java/com/wugui/datax/executor/service/command/BuildCommand.java @@ -24,6 +24,7 @@ public class BuildCommand { /** * DataX command build + * * @param tgParam * @param tmpFilePath * @param dataXPyPath @@ -47,6 +48,14 @@ public class BuildCommand { return cmdArr.toArray(new String[cmdArr.size()]); } + /** + * 构建datax运行虚拟机参数 + * + * @param tgParam + * @return {@link String} + * @author Locki + * @date 2020/9/18 + */ private static String buildDataXParam(TriggerParam tgParam) { StringBuilder doc = new StringBuilder(); String jvmParam = StringUtils.isNotBlank(tgParam.getJvmParam()) ? tgParam.getJvmParam().trim() : tgParam.getJvmParam(); @@ -56,6 +65,14 @@ public class BuildCommand { return doc.toString(); } + /** + * 构建datax增量参数 + * + * @param tgParam + * @return {@link HashMap< String, String>} + * @author Locki + * @date 2020/9/18 + */ public static HashMap buildDataXParamToMap(TriggerParam tgParam) { String partitionStr = tgParam.getPartitionInfo(); Integer incrementType = tgParam.getIncrementType(); @@ -95,6 +112,14 @@ public class BuildCommand { return null; } + /** + * 任务参数封装为map + * + * @param formatParam + * @return {@link HashMap< String, String>} + * @author Locki + * @date 2020/9/18 + */ private static HashMap getKeyValue(String formatParam) { String[] paramArr = formatParam.split(PARAMS_SYSTEM); HashMap map = new HashMap(); @@ -109,6 +134,25 @@ public class BuildCommand { return map; } + /** + * datax任务内置变量:模仿阿里云商用DataWorks/ODPS提供内置变量
+ * ${datax_bizdate} + * ${datax_biztime} + * ${datax_biz_unixtimestamp} + * + * @param + * @return {@link Map< String, String>} + * @author Locki + * @date 2020/9/18 + */ + public static Map builtInVar(){ + Map map = new HashMap(); + map.put("datax_biz_date", DateUtil.format(new Date(), "yyyy-MM-dd")); + map.put("datax_biz_time", DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss")); + map.put("datax_biz_unixtimestamp", System.currentTimeMillis() + ""); + return map; + } + private void buildPartitionCM(StringBuilder doc, String partitionStr) { if (StringUtils.isNotBlank(partitionStr)) { doc.append(SPLIT_SPACE); diff --git a/datax-executor/src/main/java/com/wugui/datax/executor/service/jobhandler/ExecutorJobHandler.java b/datax-executor/src/main/java/com/wugui/datax/executor/service/jobhandler/ExecutorJobHandler.java index 5ad80569..45699dd8 100644 --- a/datax-executor/src/main/java/com/wugui/datax/executor/service/jobhandler/ExecutorJobHandler.java +++ b/datax-executor/src/main/java/com/wugui/datax/executor/service/jobhandler/ExecutorJobHandler.java @@ -23,8 +23,7 @@ import java.util.concurrent.FutureTask; import java.util.regex.Matcher; import java.util.regex.Pattern; -import static com.wugui.datax.executor.service.command.BuildCommand.buildDataXExecutorCmd; -import static com.wugui.datax.executor.service.command.BuildCommand.buildDataXParamToMap; +import static com.wugui.datax.executor.service.command.BuildCommand.*; import static com.wugui.datax.executor.service.jobhandler.DataXConstant.DEFAULT_JSON; import static com.wugui.datax.executor.service.logparse.AnalysisStatistics.analysisStatisticsLog; @@ -55,7 +54,9 @@ public class ExecutorJobHandler extends IJobHandler { LogStatistics logStatistics = null; HashMap keyValueMap = buildDataXParamToMap(trigger); - String jobJson = replaceVariable(trigger.getJobJson(),keyValueMap); + String jobJson = replaceVariable(trigger.getJobJson(), keyValueMap); + Map buildin = builtInVar(); + jobJson = replaceVariable(jobJson, buildin); //Generate JSON temporary file tmpFilePath = generateTemJsonFile(jobJson); @@ -107,7 +108,19 @@ public class ExecutorJobHandler extends IJobHandler { } } - public static String replaceVariable(final String param,HashMap variableMap) { + /** + * 替换json变量 + * + * @param param + * @param variableMap + * @return {@link String} + * @author Locki + * @date 2020/9/18 + */ + public static String replaceVariable(final String param, Map variableMap) { + if (variableMap == null || variableMap.size() < 1) { + return param; + } Map mapping = new HashMap(); Matcher matcher = VARIABLE_PATTERN.matcher(param);