Merge branch '1.8_release_4.0.x' into temp_1.10_4.0.x_merge

kanata163 · kanata163 · commit 41b2b8a5930c · 2020-09-04T10:21:41.000+08:00
# Conflicts:
#	flinkx-rdb/flinkx-rdb-reader/src/main/java/com.dtstack.flinkx.rdb.inputformat/JdbcInputFormat.java
diff --git a/flinkx-core/src/main/java/com/dtstack/flinkx/config/DataTransferConfig.java b/flinkx-core/src/main/java/com/dtstack/flinkx/config/DataTransferConfig.java
@@ -18,8 +18,8 @@
 
 package com.dtstack.flinkx.config;
 
+import com.dtstack.flinkx.util.GsonUtil;
 import com.dtstack.flinkx.util.MapUtil;
-import com.google.gson.Gson;
 import org.apache.flink.util.Preconditions;
 
 import java.io.Reader;
@@ -110,17 +110,15 @@ private static void checkConfig(DataTransferConfig config) {
     }
 
     public static DataTransferConfig parse(String json) {
-        Gson gson = new Gson();
-        Map<String,Object> map = gson.fromJson(json, Map.class);
+        Map<String,Object> map = GsonUtil.GSON.fromJson(json, GsonUtil.gsonMapTypeToken);
         map = MapUtil.convertToHashMap(map);
         DataTransferConfig config = new DataTransferConfig(map);
         checkConfig(config);
         return config;
     }
 
     public static DataTransferConfig parse(Reader reader) {
-        Gson gson = new Gson();
-        DataTransferConfig config = gson.fromJson(reader, DataTransferConfig.class);
+        DataTransferConfig config = GsonUtil.GSON.fromJson(reader, DataTransferConfig.class);
         checkConfig(config);
         return config;
     }
diff --git a/flinkx-core/src/main/java/com/dtstack/flinkx/enums/ColumnType.java b/flinkx-core/src/main/java/com/dtstack/flinkx/enums/ColumnType.java
@@ -35,12 +35,12 @@ public enum ColumnType {
     /**
      * string type
      */
-    STRING, VARCHAR,VARCHAR2, CHAR,NVARCHAR,TEXT,KEYWORD,BINARY,
+    STRING, VARCHAR, VARCHAR2, CHAR, NVARCHAR, TEXT, KEYWORD, BINARY,
 
     /**
      * number type
      */
-    INT, INT32, MEDIUMINT, TINYINT, DATETIME, SMALLINT, BIGINT,LONG, INT64 ,SHORT,INTEGER,
+    INT, INT32, MEDIUMINT, TINYINT, DATETIME, SMALLINT, BIGINT, LONG, INT64 , SHORT, INTEGER, NUMBER,
 
     /**
      * double type
@@ -51,19 +51,19 @@ public enum ColumnType {
     /**
      * date type
      */
-    DATE, TIMESTAMP,TIME,
-    DECIMAL,YEAR,BIT;
+    DATE, TIMESTAMP, TIME,
+    DECIMAL, YEAR, BIT;
 
     public static List<ColumnType> TIME_TYPE = Arrays.asList(
-            DATE,DATETIME,TIME,TIMESTAMP
+            DATE, DATETIME, TIME, TIMESTAMP
     );
 
     public static List<ColumnType> NUMBER_TYPE = Arrays.asList(
-            INT,INTEGER,MEDIUMINT,TINYINT,SMALLINT, BIGINT,LONG,SHORT,DOUBLE, FLOAT,DECIMAL
+            INT, INTEGER, MEDIUMINT, TINYINT, SMALLINT, BIGINT, LONG, SHORT, DOUBLE, FLOAT, DECIMAL, NUMBER
     );
 
     public static List<ColumnType> STRING_TYPE = Arrays.asList(
-            STRING, VARCHAR,VARCHAR2, CHAR,NVARCHAR,TEXT,KEYWORD,BINARY
+            STRING, VARCHAR, VARCHAR2, CHAR, NVARCHAR, TEXT, KEYWORD, BINARY
     );
 
     public static ColumnType fromString(String type) {
diff --git a/flinkx-core/src/main/java/com/dtstack/flinkx/inputformat/BaseRichInputFormat.java b/flinkx-core/src/main/java/com/dtstack/flinkx/inputformat/BaseRichInputFormat.java
@@ -188,7 +188,7 @@ private void showConfig(){
                 writerParameter.put(KEY_PASSWORD, KEY_CONFUSED_PASSWORD);
             }
         }
-    LOG.info("configInfo : {}", GsonUtil.GSON.toJson(map));
+        LOG.info("configInfo : \n{}", GsonUtil.GSON.toJson(map));
     }
 
     private void checkIfCreateSplitFailed(InputSplit inputSplit){
diff --git a/flinkx-core/src/main/java/com/dtstack/flinkx/util/DateUtil.java b/flinkx-core/src/main/java/com/dtstack/flinkx/util/DateUtil.java
@@ -18,6 +18,8 @@
 
 package com.dtstack.flinkx.util;
 
+import org.apache.commons.lang3.StringUtils;
+
 import java.sql.Timestamp;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
@@ -26,6 +28,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.TimeZone;
+import java.util.regex.Pattern;
 
 /**
  * Date Utilities
@@ -250,4 +253,67 @@ public static SimpleDateFormat buildDateFormatter(String timeFormat){
         sdf.setTimeZone(TimeZone.getTimeZone(TIME_ZONE));
         return sdf;
     }
+
+    /**
+     * 常规自动日期格式识别
+     * @param str 时间字符串
+     * @return String DateFormat字符串如：yyyy-MM-dd HH:mm:ss
+     */
+    public static String getDateFormat(String str) {
+        if(StringUtils.isBlank(str)){
+            return null;
+        }
+        boolean year = false;
+        Pattern pattern = Pattern.compile("^[-\\+]?[\\d]*$");
+        if(pattern.matcher(str.substring(0, 4)).matches()) {
+            year = true;
+        }
+        StringBuilder sb = new StringBuilder();
+        int index = 0;
+        if(!year) {
+            if(str.contains("月") || str.contains("-") || str.contains("/")) {
+                if(Character.isDigit(str.charAt(0))) {
+                    index = 1;
+                }
+            }else {
+                index = 3;
+            }
+        }
+        for (int i = 0; i < str.length(); i++) {
+            char chr = str.charAt(i);
+            if(Character.isDigit(chr)) {
+                if(index==0) {
+                    sb.append("y");
+                }
+                if(index==1) {
+                    sb.append("M");
+                }
+                if(index==2) {
+                    sb.append("d");
+                }
+                if(index==3) {
+                    sb.append("H");
+                }
+                if(index==4) {
+                    sb.append("m");
+                }
+                if(index==5) {
+                    sb.append("s");
+                }
+                if(index==6) {
+                    sb.append("S");
+                }
+            }else {
+                if(i>0) {
+                    char lastChar = str.charAt(i-1);
+                    if(Character.isDigit(lastChar)) {
+                        index++;
+                    }
+                }
+                sb.append(chr);
+            }
+        }
+        return sb.toString();
+    }
+
 }
diff --git a/flinkx-core/src/main/java/com/dtstack/flinkx/util/GsonUtil.java b/flinkx-core/src/main/java/com/dtstack/flinkx/util/GsonUtil.java
@@ -50,7 +50,9 @@ public class GsonUtil {
 
     @SuppressWarnings("unchecked")
     private static Gson getGson() {
-        GSON = new GsonBuilder().create();
+        GSON = new GsonBuilder()
+                .setPrettyPrinting()
+                .create();
         try {
             Field factories = Gson.class.getDeclaredField("factories");
             factories.setAccessible(true);
diff --git a/flinkx-core/src/main/java/com/dtstack/flinkx/util/ResultPrintUtil.java b/flinkx-core/src/main/java/com/dtstack/flinkx/util/ResultPrintUtil.java
@@ -49,7 +49,8 @@ public static void printResult(JobExecutionResult result){
         }
         maxLength += 5;
 
-        StringBuilder builder = new StringBuilder();
+        StringBuilder builder = new StringBuilder(128);
+        builder.append("\n*********************************************\n");
         for (int i = 0; i < names.size(); i++) {
             String name = names.get(i);
             builder.append(name + StringUtils.repeat(" ", maxLength - name.length()));
@@ -59,9 +60,7 @@ public static void printResult(JobExecutionResult result){
                 builder.append("\n");
             }
         }
-
-        LOG.info("---------------------------------");
+        builder.append("\n*********************************************\n");
         LOG.info(builder.toString());
-        LOG.info("---------------------------------");
     }
 }
diff --git a/flinkx-core/src/main/java/com/dtstack/flinkx/util/StringUtil.java b/flinkx-core/src/main/java/com/dtstack/flinkx/util/StringUtil.java
@@ -20,11 +20,14 @@
 
 import com.dtstack.flinkx.enums.ColumnType;
 import com.dtstack.flinkx.exception.WriteRecordException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.time.DateUtils;
 import org.apache.flink.types.Row;
 
 import java.math.BigDecimal;
 import java.sql.Date;
 import java.sql.Timestamp;
+import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.List;
@@ -296,4 +299,26 @@ public static List<String> splitIgnoreQuota(String str, char delimiter){
 
         return tokensList;
     }
+
+    /**
+     * 字符串转换成对应时间戳字符串
+     * @param location
+     * @return
+     */
+    public static String stringToTimestampStr(String location, ColumnType type){
+        //若为空字符串或本身就是时间戳则不需要转换
+        if(StringUtils.isBlank(location) || StringUtils.isNumeric(location)){
+            return location;
+        }
+        try {
+            switch (type) {
+                case TIMESTAMP: return String.valueOf(Timestamp.valueOf(location).getTime());
+                case DATE: return String.valueOf(DateUtils.parseDate(location, DateUtil.getDateFormat(location)).getTime());
+                default: return location;
+            }
+        }catch (ParseException e){
+            String message = String.format("cannot transform 【%s】to 【%s】, e = %s", location, type, ExceptionUtil.getErrorMessage(e));
+            throw new RuntimeException(message);
+        }
+    }
 }
diff --git a/flinkx-rdb/flinkx-rdb-reader/src/main/java/com.dtstack.flinkx.rdb.inputformat/JdbcInputFormat.java b/flinkx-rdb/flinkx-rdb-reader/src/main/java/com.dtstack.flinkx.rdb.inputformat/JdbcInputFormat.java

Original file line number	Diff line number	Diff line change
`@@ -188,7 +188,7 @@ private void showConfig(){`
`188`	`188`	`writerParameter.put(KEY_PASSWORD, KEY_CONFUSED_PASSWORD);`
`189`	`189`	`}`
`190`	`190`	`}`
`191`		`- LOG.info("configInfo : {}", GsonUtil.GSON.toJson(map));`
	`191`	`+ LOG.info("configInfo : \n{}", GsonUtil.GSON.toJson(map));`
`192`	`192`	`}`
`193`	`193`
`194`	`194`	`private void checkIfCreateSplitFailed(InputSplit inputSplit){`
Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,8 @@ public static void printResult(JobExecutionResult result){`
`49`	`49`	`}`
`50`	`50`	`maxLength += 5;`
`51`	`51`
`52`		`- StringBuilder builder = new StringBuilder();`
	`52`	`+ StringBuilder builder = new StringBuilder(128);`
	`53`	`+ builder.append("\n*********************************************\n");`
`53`	`54`	`for (int i = 0; i < names.size(); i++) {`
`54`	`55`	`String name = names.get(i);`
`55`	`56`	`builder.append(name + StringUtils.repeat(" ", maxLength - name.length()));`
`@@ -59,9 +60,7 @@ public static void printResult(JobExecutionResult result){`
`59`	`60`	`builder.append("\n");`
`60`	`61`	`}`
`61`	`62`	`}`
`62`		`-`
`63`		`- LOG.info("---------------------------------");`
	`63`	`+ builder.append("\n*********************************************\n");`
`64`	`64`	`LOG.info(builder.toString());`
`65`		`- LOG.info("---------------------------------");`
`66`	`65`	`}`
`67`	`66`	`}`