[merge]resolve conflits from merge.

Author: HiLany

docs/plugin/elasticsearch6Sink.md

@@ -9,7 +9,7 @@ CREATE TABLE tableName(
     cluster='clusterName',
     esType ='esType',
     index ='index',
-    id ='num[,num]',
+    id ='num[,num]'(id = 'field[,field]'),
     authMesh = 'true',
     userName = 'userName',
     password = 'password',
@@ -35,7 +35,7 @@ CREATE TABLE tableName(
 |cluster | ES 集群名称 |是||
 |index | 选择的ES上的index名称|是||
 |esType | 选择ES上的type名称|是||
-|id | 生成id的规则(当前是根据指定的字段pos获取字段信息,拼接生成id;|否||
+|id | 生成id的规则(当前是根据指定的字段名称(或者字段position)获取字段信息,拼接生成id)|否||
 | |若id为空字符串或索引都超出范围，则随机生成id值)|||
 |authMesh | 是否进行用户名密码认证 | 否 | false|
 |userName | 用户名 | 否，authMesh='true'时为必填 ||
@@ -73,7 +73,7 @@ CREATE TABLE MyResult(
     estype ='external',
     cluster ='docker-cluster',
     index ='myresult',
-    id ='1',
+    id ='pv',
     updateMode ='append',
     parallelism ='1'
  );

docs/plugin/elasticsearchSink.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'
  )
 ```
@@ -32,7 +32,7 @@ CREATE TABLE tableName(
 |cluster | ES 集群名称 |是||
 |index | 选择的ES上的index名称|是||
 |estype | 选择ES上的type名称|是||
-|id | 生成id的规则(当前是根据指定的字段pos获取字段信息,拼接生成id)|是||
+|id | 生成id的规则(当前是根据指定的字段名称(或者字段position)获取字段信息,拼接生成id)|否||
 |authMesh | 是否进行用户名密码认证 | 否 | false|
 |userName | 用户名 | 否，authMesh='true'时为必填 ||
 |password | 密码 | 否，authMesh='true'时为必填 ||
@@ -68,7 +68,7 @@ CREATE TABLE MyResult(
     estype ='external',
     cluster ='docker-cluster',
     index ='myresult',
-    id ='1',
+    id ='pv',
     updateMode ='append',
     parallelism ='1'
  );

elasticsearch5/elasticsearch5-sink/pom.xml

@@ -16,13 +16,13 @@
         <dependency>
             <groupId>org.elasticsearch.client</groupId>
             <artifactId>transport</artifactId>
-            <version>5.3.3</version>
+            <version>${elasticsearch.version}</version>
         </dependency>
 
         <dependency>
             <groupId>org.elasticsearch.client</groupId>
             <artifactId>x-pack-transport</artifactId>
-            <version>5.3.3</version>
+            <version>${elasticsearch.version}</version>
         </dependency>
 
         <!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-log4j12 -->

elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/CustomerSinkFunc.java

@@ -53,23 +53,24 @@ public class CustomerSinkFunc implements ElasticsearchSinkFunction<Tuple2> {
 
     private String type;
 
-    private List<Integer> idFieldIndexList;
+    private List<Object> ids;
+
+    // true means generation doc's id by position "1[,1]"
+    private boolean usePosition;
 
     private List<String> fieldNames;
 
     private List<String> fieldTypes;
 
     public transient Counter outRecords;
 
-    /** 默认分隔符为'_' */
-    private char sp = '_';
-
-    public CustomerSinkFunc(String index, String type, List<String> fieldNames, List<String> fieldTypes, List<Integer> idFieldIndexes){
+    public CustomerSinkFunc(String index, String type, List<String> fieldNames, List<String> fieldTypes, List<Object> ids, boolean usePosition) {
         this.index = index;
         this.type = type;
         this.fieldNames = fieldNames;
         this.fieldTypes = fieldTypes;
-        this.idFieldIndexList = idFieldIndexes;
+        this.ids = ids;
+        this.usePosition = usePosition;
     }
 
     @Override
@@ -95,21 +96,31 @@ public void setOutRecords(Counter outRecords) {
     }
 
     private IndexRequest createIndexRequest(Row element) {
-        String idFieldStr = "";
-        if (null != idFieldIndexList) {
-            // index start at 1,
-            idFieldStr = idFieldIndexList.stream()
-                    .filter(index -> index > 0 && index <= element.getArity())
-                    .map(index -> element.getField(index - 1).toString())
-                    .collect(Collectors.joining(ID_VALUE_SPLIT));
-        }
 
         Map<String, Object> dataMap = EsUtil.rowToJsonMap(element,fieldNames,fieldTypes);
         int length = Math.min(element.getArity(), fieldNames.size());
         for(int i=0; i<length; i++){
             dataMap.put(fieldNames.get(i), element.getField(i));
         }
 
+        String idFieldStr = "";
+        if (null != ids) {
+            if (!usePosition) {
+                idFieldStr = ids.stream()
+                        .map(filedName -> (String) filedName)
+                        .map(filedName -> dataMap.get(filedName).toString())
+                        .collect(Collectors.joining(ID_VALUE_SPLIT));
+            } else {
+                // compatible old version of generate doc's id
+                // index start at 1,
+                idFieldStr = ids.stream()
+                        .map(index -> (Integer) index)
+                        .filter(index -> index > 0 && index <= element.getArity())
+                        .map(index -> element.getField( index - 1).toString())
+                        .collect(Collectors.joining(ID_VALUE_SPLIT));
+            }
+        }
+
         if (StringUtils.isEmpty(idFieldStr)) {
             return Requests.indexRequest()
                     .index(index)

elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ElasticsearchSink.java

@@ -21,6 +21,7 @@
 package com.dtstack.flink.sql.sink.elasticsearch;
 
 import com.dtstack.flink.sql.table.AbstractTargetTableInfo;
+import com.google.common.base.Preconditions;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -46,6 +47,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 /**
  * table output elastic5plugin
@@ -68,7 +70,10 @@ public class ElasticsearchSink implements RetractStreamTableSink<Row>, IStreamSi
 
     private String type = "";
 
-    private List<Integer> idIndexList;
+    private List<Object> ids;
+
+    // true means generation doc's id by position "1[,1]"
+    private boolean usePosition = false;
 
     protected String[] fieldNames;
 
@@ -113,6 +118,14 @@ public TypeInformation<?>[] getFieldTypes() {
 
     private RichSinkFunction createEsSinkFunction(){
 
+        // check whether id fields is exists in columns if not use position to generate doc's id
+        if (!usePosition
+                && ids != null
+                && ids.size() != 0) {
+            List<String> filedNamesLists = Arrays.asList(fieldNames);
+            Preconditions.checkState(filedNamesLists.containsAll(ids), "elasticsearch6 type of id %s is should be exists in columns %s.", ids, filedNamesLists);
+        }
+        CustomerSinkFunc customerSinkFunc = new CustomerSinkFunc(index, type, Arrays.asList(fieldNames), Arrays.asList(columnTypes), ids, usePosition);
 
         Map<String, String> userConfig = new HashMap<>();
         userConfig.put("cluster.name", clusterName);
@@ -142,8 +155,6 @@ private RichSinkFunction createEsSinkFunction(){
             userConfig.put("xpack.security.user", authPassword);
         }
 
-        CustomerSinkFunc customerSinkFunc = new CustomerSinkFunc(index, type, Arrays.asList(fieldNames), Arrays.asList(columnTypes), idIndexList);
-
         return new MetricElasticsearchSink(userConfig, transports, customerSinkFunc, esTableInfo);
     }
 
@@ -179,15 +190,21 @@ public ElasticsearchSink genStreamSink(AbstractTargetTableInfo targetTableInfo)
         type = elasticsearchTableInfo.getEsType();
         String id = elasticsearchTableInfo.getId();
         String[] idField = StringUtils.split(id, ",");
-        idIndexList = new ArrayList<>();
         registerTableName = elasticsearchTableInfo.getName();
         parallelism = Objects.isNull(elasticsearchTableInfo.getParallelism()) ?
                 parallelism : elasticsearchTableInfo.getParallelism();
 
         for(int i = 0; i < idField.length; ++i) {
-            idIndexList.add(Integer.valueOf(idField[i]));
+            if (!EsUtil.checkWhetherUsePosition(id)) {
+                ids = Arrays.stream(org.apache.commons.lang.StringUtils.split(id, ",")).map(String::valueOf).collect(Collectors.toList());
+            } else {
+                //compatible old version of generate doc' id
+                usePosition = true;
+                ids = Arrays.stream(org.apache.commons.lang.StringUtils.split(id, ",")).map(Integer::valueOf).collect(Collectors.toList());
+            }
         }
 
+
         columnTypes = elasticsearchTableInfo.getFieldTypes();
 
         return this;

elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/EsUtil.java

@@ -18,6 +18,8 @@
 
 package com.dtstack.flink.sql.sink.elasticsearch;
 
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.math.NumberUtils;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Preconditions;
 
@@ -62,5 +64,23 @@ public static Map<String, Object> rowToJsonMap(Row row, List<String> fields, Lis
         return jsonMap;
     }
 
+    /**
+     * check whether use position to generation doc's id
+     * eg : |1,2,3 -> true
+     *      |id,name,addr -> false
+     * @param ids
+     * @return
+     */
+    public static boolean checkWhetherUsePosition(String ids) {
+        boolean flag = true;
+        for( String id : StringUtils.split(ids, ",")) {
+            if (!NumberUtils.isNumber(id)) {
+                flag= false;
+                break;
+            }
+        }
+        return flag;
+    }
+
 
 }

elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/MetricElasticsearchSink.java

@@ -60,20 +60,6 @@ public void open(Configuration parameters) throws Exception {
         initMetric();
     }
 
-    /*public void setXPackTransportClient() throws Exception {
-        String authPassword = esTableInfo.getUserName() + ":" + esTableInfo.getPassword();
-        Settings settings = Settings.builder().put(userConfig).put("xpack.security.user", authPassword).build();
-        Class clz = Class.forName("org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase");
-        Field clientField = clz.getDeclaredField("client");
-        clientField.setAccessible(true);
-        PreBuiltXPackTransportClient transportClient = new PreBuiltXPackTransportClient(settings);
-        for (TransportAddress transport : ElasticsearchUtils.convertInetSocketAddresses(transportAddresses)) {
-            transportClient.addTransportAddress(transport);
-        }
-
-        clientField.set(this, transportClient);
-    }*/
-
     public void initMetric() {
         Counter counter = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_OUT);
         customerSinkFunc.setOutRecords(counter);

elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/table/ElasticsearchTableInfo.java

@@ -134,12 +134,6 @@ public boolean check() {
         Preconditions.checkNotNull(esType, "elasticsearch type of type is required");
         Preconditions.checkNotNull(clusterName, "elasticsearch type of clusterName is required");
 
-        if (!StringUtils.isEmpty(id)) {
-            Arrays.stream(StringUtils.split(id, ",")).forEach(number -> {
-                Preconditions.checkArgument(NumberUtils.isNumber(number), "id must be a numeric type");
-            });
-        }
-
         if (isAuthMesh()) {
             Preconditions.checkNotNull(userName, "elasticsearch type of userName is required");
             Preconditions.checkNotNull(password, "elasticsearch type of password is required");

elasticsearch5/elasticsearch5-sink/src/test/java/com/dtstack/flink/sql/sink/elasticsearch/CustomerSinkFuncTest.java

@@ -32,13 +32,13 @@ public class CustomerSinkFuncTest {
     String type;
     List<String> fieldNames = Collections.singletonList("name.pv");
     List<String> fieldTypes = Arrays.asList("varchar", "varchar");
-    List<Integer> idFieldIndexes = Collections.singletonList(1);
+    List<Object> idFieldIndexes = Collections.singletonList(1);
     Tuple2<Boolean, Row> tuple2 = new Tuple2<>();
 
     @Before
     public void setUp() {
         MockitoAnnotations.initMocks(this);
-        customerSinkFunc = new CustomerSinkFunc(index, type, fieldNames, fieldTypes, idFieldIndexes);
+        customerSinkFunc = new CustomerSinkFunc(index, type, fieldNames, fieldTypes, idFieldIndexes, true);
         customerSinkFunc.setOutRecords(outRecords);
         tuple2.setField(true, 0);
         Row row = new Row(1);

elasticsearch5/elasticsearch5-sink/src/test/java/com/dtstack/flink/sql/sink/elasticsearch/MetricElasticsearchSinkTest.java

@@ -11,7 +11,6 @@
 import org.junit.Test;
 import org.mockito.MockitoAnnotations;
 
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;