Merge pull request #416 from DTStack/feat_1.10_415

支持elasticsearch7 sink以及side，并完善其文档
支持脏数据管理
支持file source
impla插件批量写入失败时改为单条insert
支持http sink

Author: FlechazoW

README.md

@@ -41,5 +41,7 @@ FlinkStreamSQL
 
 ## 如何贡献FlinkStreamSQL
 
+[pr规范](docs/pr.md)
+
 ## License
 FlinkStreamSQL is under the Apache 2.0 license. See the [LICENSE](http://www.apache.org/licenses/LICENSE-2.0) file for details.    

core/src/main/java/com/dtstack/flink/sql/enums/ColumnType.java

@@ -102,6 +102,10 @@ public enum ColumnType {
      * timestamp
      */
     TIMESTAMP,
+    /**
+     * time eg: 23:59:59
+     */
+    TIME,
     /**
      * decimal
      */

core/src/main/java/com/dtstack/flink/sql/exception/ExceptionTrace.java

@@ -20,21 +20,4 @@ public static String traceOriginalCause(Throwable e) {
         }
         return errorMsg;
     }
-
-    /**
-     * 根据异常的种类来判断是否需要强制跳过Flink的重启{@link SuppressRestartsException}
-      * @param e exception
-     * @param errorMsg 需要抛出的异常信息
-     */
-    public static void dealExceptionWithSuppressStart(Exception e, String errorMsg) {
-        if (e instanceof SuppressRestartsException) {
-            throw new SuppressRestartsException(
-                new Throwable(
-                    errorMsg
-                )
-            );
-        } else {
-            throw new RuntimeException(errorMsg);
-        }
-    }
 }

core/src/main/java/com/dtstack/flink/sql/side/AbstractSideTableInfo.java

@@ -193,6 +193,10 @@ public List<PredicateInfo> getPredicateInfoes() {
         return predicateInfoes;
     }
 
+    public void addPredicateInfo(PredicateInfo predicateInfo) {
+        this.predicateInfoes.add(predicateInfo);
+    }
+
     public Long getAsyncFailMaxNum(Long defaultValue) {
         return Objects.isNull(asyncFailMaxNum) ? defaultValue : asyncFailMaxNum;
     }

core/src/main/java/com/dtstack/flink/sql/side/BaseSideInfo.java

@@ -137,12 +137,14 @@ public void dealOneEqualCon(SqlNode sqlNode, String sideTableName) {
         if (leftNode.getKind() == SqlKind.LITERAL) {
             evalConstantEquation(
                 (SqlLiteral) leftNode,
-                (SqlIdentifier) rightNode
+                (SqlIdentifier) rightNode,
+                sqlNode.getKind()
             );
         } else if (rightNode.getKind() == SqlKind.LITERAL) {
             evalConstantEquation(
                 (SqlLiteral) rightNode,
-                (SqlIdentifier) leftNode
+                (SqlIdentifier) leftNode,
+                sqlNode.getKind()
             );
         } else {
             SqlIdentifier left = (SqlIdentifier) leftNode;
@@ -179,20 +181,19 @@ private void evalEquation(SqlIdentifier left, SqlIdentifier right, String sideTa
      * @param literal
      * @param identifier
      */
-    private void evalConstantEquation(SqlLiteral literal, SqlIdentifier identifier) {
+    private void evalConstantEquation(SqlLiteral literal, SqlIdentifier identifier, SqlKind sqlKind) {
         String tableName = identifier.getComponent(0).getSimple();
         checkSupport(identifier);
         String fieldName = identifier.getComponent(1).getSimple();
         Object constant = literal.getValue();
-        List<PredicateInfo> predicateInfos = sideTableInfo.getPredicateInfoes();
         PredicateInfo predicate = PredicateInfo.builder()
-            .setOperatorName("=")
-            .setOperatorKind("EQUALS")
+            .setOperatorName(sqlKind.sql)
+            .setOperatorKind(sqlKind.name())
             .setOwnerTable(tableName)
             .setFieldName(fieldName)
             .setCondition(constant.toString())
             .build();
-        predicateInfos.add(predicate);
+        sideTableInfo.addPredicateInfo(predicate);
     }
 
     private void checkSupport(SqlIdentifier identifier) {

core/src/main/java/com/dtstack/flink/sql/util/DtStringUtil.java

@@ -21,15 +21,17 @@
 package com.dtstack.flink.sql.util;
 
 import com.dtstack.flink.sql.enums.ColumnType;
-import org.apache.commons.lang3.StringUtils;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Strings;
 import com.google.common.collect.Maps;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import java.sql.Timestamp;
+import org.apache.commons.lang3.StringUtils;
+
 import java.math.BigDecimal;
+import java.sql.Timestamp;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -241,6 +243,9 @@ public static String col2string(Object column, String type) {
             case DATE:
                 result = DateUtil.dateToString((java.util.Date)column);
                 break;
+            case TIME:
+                result = DateUtil.getTimeFromStr(String.valueOf(column));
+                break;
             case TIMESTAMP:
                 result = DateUtil.timestampToString((java.util.Date)column);
                 break;
@@ -411,4 +416,14 @@ public static String removeStartAndEndQuota(String str) {
         String removeStart = StringUtils.removeStart(str, "'");
         return StringUtils.removeEnd(removeStart, "'");
     }
+
+    /**
+     * 判断当前对象是null 还是空格
+     *
+     * @param obj 需要判断的对象
+     * @return 返回true 如果对象是空格或者为null
+     */
+    public static boolean isEmptyOrNull(Object obj) {
+        return Objects.isNull(obj) || obj.toString().isEmpty();
+    }
 }

core/src/main/java/com/dtstack/flink/sql/util/KrbUtils.java

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.util;
 
+import com.esotericsoftware.minlog.Log;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authentication.util.KerberosName;

docs/plugin/elasticsearchSink.md renamed to docs/plugin/elasticsearch5Sink.md

@@ -9,7 +9,7 @@ CREATE TABLE tableName(
     cluster='clusterName',
     estype ='esType',
     index ='index',
-    id ='num[,num]',
+    id ='num[,num]'(id = 'field[,field]'),
     parallelism ='1'
  )
 ```
@@ -27,19 +27,19 @@ CREATE TABLE tableName(
 ## 4.参数：
 |参数名称|含义|是否必填|默认值|
 |----|---|---|----|
-|type|表明 输出表类型[mysq|hbase|elasticsearch]|是||
-|address | 连接ES Transport地址(tcp地址)|是||
+|type|表明 输出表类型[mysq|hbase|elasticsearch]|是|elasticsearch|
+|address | 连接ES Transport地址(tcp地址)|是|9300|
 |cluster | ES 集群名称 |是||
 |index | 选择的ES上的index名称|是||
 |estype | 选择ES上的type名称|是||
-|id | 生成id的规则(当前是根据指定的字段pos获取字段信息,拼接生成id)|是||
-|authMesh | 是否进行用户名密码认证 | 否 | false|
+|id | 生成id的规则(当前是根据指定的字段名称(或者字段position)获取字段信息,拼接生成id)|否||
+|authMesh | 是否进行用户名密码认证(xpack认证) | 否 | false|
 |userName | 用户名 | 否，authMesh='true'时为必填 ||
 |password | 密码 | 否，authMesh='true'时为必填 ||
 |parallelism | 并行度设置|否|1|
 
 ## 5.样例：
-```
+```sql
 CREATE TABLE MyTable(
     channel varchar,
     pv varchar
@@ -61,14 +61,15 @@ CREATE TABLE MyResult(
     channel varchar
  )WITH(
     type ='elasticsearch',
-    address ='172.16.8.193:9200',
+    address ='172.16.8.193:9300',
     authMesh='true',
     username='elastic',
     password='abc123',
     estype ='external',
     cluster ='docker-cluster',
     index ='myresult',
-    id ='1',
+    id ='pv',
+--  id = '1' # 在支持position方式和属性名方式
     updateMode ='append',
     parallelism ='1'
  );

docs/plugin/elasticsearch6Side.md

@@ -1,5 +1,4 @@
-
-## 1.格式：
+## 1.格式
 ```
  CREATE TABLE tableName(
      colName cloType,
@@ -24,45 +23,49 @@
 ```
 
 # 2.支持版本
- elasticsearch 6.8.6
+
+ elasticsearch `6.x`
 
 ## 3.表结构定义
-  
- |参数名称|含义|
- |----|---|
- | tableName | elasticsearch表名称|
- | colName | 列名称|
- | colType | 列类型 [colType支持的类型](../colType.md)|
- | PERIOD FOR SYSTEM_TIME | 关键字表明该定义的表为维表信息|
- | PRIMARY KEY(keyInfo) | 维表主键定义;多个列之间用逗号隔开|
+   
+|参数名称|含义|
+|---|---|
+| tableName | elasticsearch表名称|
+| colName | 列名称|
+| colType | 列类型 [colType支持的类型](../colType.md)|
+| PERIOD FOR SYSTEM_TIME | 关键字表明该定义的表为维表信息|
+| PRIMARY KEY(keyInfo) | 维表主键定义;多个列之间用逗号隔开|
 
 ## 4.参数
 
-  |参数名称|含义|是否必填|默认值|
-  |----|---|---|----|
-  type|表明 输出表类型[elasticsearch6]|是||
-  |address | 连接ES Transport地址(tcp地址)|是||
-  |cluster | ES 集群名称 |是||
-  |index | 选择的ES上的index名称|否||
-  |esType | 选择ES上的type名称|否||
-  |authMesh | 是否进行用户名密码认证 | 否 | false|
-  |userName | 用户名 | 否，authMesh='true'时为必填 ||
-  |password | 密码 | 否，authMesh='true'时为必填 ||
-  | cache | 维表缓存策略(NONE/LRU)|否|NONE|
-  | partitionedJoin | 是否在維表join之前先根据 設定的key 做一次keyby操作(可以減少维表的数据缓存量)|否|false|
-  |parallelism | 并行度设置|否|1|
+|参数名称|含义|是否必填|默认值|
+|----|---|---|----|
+|type|表明 输出表类型[elasticsearch6]|是||
+|address | 连接ES Http地址|是||
+|cluster | ES 集群名称 |是||
+|index | 选择的ES上的index名称|否||
+|esType | 选择ES上的type名称|否||
+|authMesh | 是否进行用户名密码认证 | 否 | false|
+|userName | 用户名 | 否，authMesh='true'时为必填 ||
+|password | 密码 | 否，authMesh='true'时为必填 ||
+| cache | 维表缓存策略(NONE/LRU)|否|NONE|
+| partitionedJoin | 是否在維表join之前先根据 設定的key 做一次keyby操作(可以減少维表的数据缓存量)|否|false|
+|parallelism | 并行度设置|否|1|
 
-  ----------
-  > 缓存策略
-  * NONE: 不做内存缓存
-  * LRU:
+
+----------
+> 缓存策略
+
+* NONE: 不做内存缓存
+* LRU:
     * cacheSize: 缓存的条目数量
     * cacheTTLMs:缓存的过期时间(ms)
     * cacheMode: (unordered|ordered)异步加载是有序还是无序,默认有序。
     * asyncCapacity:异步请求容量，默认1000
     * asyncTimeout：异步请求超时时间，默认10000毫秒
 
 ## 5.样例
+
 ```
 create table sideTable(
     channel varchar,
@@ -71,7 +74,7 @@ create table sideTable(
     PERIOD FOR SYSTEM_TIME
  )WITH(
     type ='elasticsearch6',
-    address ='172.16.10.47:9500',
+    address ='localhost:9200',
     cluster='es_47_menghan',
     estype ='type1',
     index ='xc_es_test',

docs/plugin/elasticsearch6Sink.md

@@ -9,15 +9,15 @@ CREATE TABLE tableName(
     cluster='clusterName',
     esType ='esType',
     index ='index',
-    id ='num[,num]',
+    id ='num[,num]'(id = 'field[,field]'),
     authMesh = 'true',
     userName = 'userName',
     password = 'password',
     parallelism ='1'
  )
 ```
 ## 2.支持的版本
-   elasticsearch 6.8.6
+   elasticsearch `6.x`
 
 ## 3.表结构定义
 
@@ -31,11 +31,11 @@ CREATE TABLE tableName(
 |参数名称|含义|是否必填|默认值|
 |----|---|---|----|
 |type|表明 输出表类型[elasticsearch6]|是||
-|address | 连接ES Transport地址(tcp地址)|是||
+|address | 连接ES Http地址|是||
 |cluster | ES 集群名称 |是||
 |index | 选择的ES上的index名称|是||
 |esType | 选择ES上的type名称|是||
-|id | 生成id的规则(当前是根据指定的字段pos获取字段信息,拼接生成id;|否||
+|id | 生成id的规则(当前是根据指定的字段名称(或者字段position)获取字段信息,拼接生成id)|否||
 | |若id为空字符串或索引都超出范围，则随机生成id值)|||
 |authMesh | 是否进行用户名密码认证 | 否 | false|
 |userName | 用户名 | 否，authMesh='true'时为必填 ||
@@ -73,8 +73,8 @@ CREATE TABLE MyResult(
     estype ='external',
     cluster ='docker-cluster',
     index ='myresult',
-    id ='1',
-    updateMode ='append',
+    id ='pv',
+--  id = '1' # 在支持position方式和属性名方式
     parallelism ='1'
  );