[feat-269][elasticsearch6][sink]compatible old version of generate doc's id

Author: HiLany

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

@@ -53,7 +53,10 @@ public class CustomerSinkFunc implements ElasticsearchSinkFunction<Tuple2> {
 
     private String type;
 
-    private List<String> idFiledNames;
+    private List<Object> ids;
+
+    // true means generation doc's id by position "1[,1]"
+    private boolean usePosition;
 
     private List<String> fieldNames;
 
@@ -63,12 +66,13 @@ public class CustomerSinkFunc implements ElasticsearchSinkFunction<Tuple2> {
 
     private transient Counter outDirtyRecords;
 
-    public CustomerSinkFunc(String index, String type, List<String> fieldNames, List<String> fieldTypes, List<String> idFiledNames) {
+    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.idFiledNames = idFiledNames;
+        this.ids = ids;
+        this.usePosition = usePosition;
     }
 
     @Override
@@ -111,10 +115,21 @@ private IndexRequest createIndexRequest(Row element) {
         }
 
         String idFieldStr = "";
-        if (null != idFiledNames) {
-            idFieldStr = idFiledNames.stream()
-                    .map(idFiledName -> dataMap.get(idFiledName).toString())
-                    .collect(Collectors.joining(ID_VALUE_SPLIT));
+        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)) {

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

@@ -62,7 +62,10 @@ public class ElasticsearchSink implements RetractStreamTableSink<Row>, IStreamSi
 
     private String type = "";
 
-    private List<String> idFiledNames;
+    private List<Object> ids;
+
+    // true means generation doc's id by position "1[,1]"
+    private boolean usePosition = false;
 
     protected String[] fieldNames;
 
@@ -107,10 +110,12 @@ public TypeInformation<?>[] getFieldTypes() {
 
     private RichSinkFunction createEsSinkFunction() {
 
-        // check whether id fields is exists in columns
-        List<String> filedNamesLists = Arrays.asList(fieldNames);
-        Preconditions.checkState(filedNamesLists.containsAll(idFiledNames), "elasticsearch6 type of id %s is should be exists in columns %s.", idFiledNames, filedNamesLists);
-        CustomerSinkFunc customerSinkFunc = new CustomerSinkFunc(index, type, filedNamesLists, Arrays.asList(columnTypes), idFiledNames);
+        // check whether id fields is exists in columns if not use position to generate doc's id
+        if (!usePosition) {
+            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 = Maps.newHashMap();
         userConfig.put("cluster.name", clusterName);
@@ -156,8 +161,15 @@ public ElasticsearchSink genStreamSink(AbstractTargetTableInfo targetTableInfo)
         registerTableName = esTableInfo.getName();
         parallelism = Objects.isNull(esTableInfo.getParallelism()) ? parallelism : esTableInfo.getParallelism();
 
+        // check mode of generate doc's id
         if (!StringUtils.isEmpty(id)) {
-            idFiledNames = Arrays.stream(StringUtils.split(id, ",")).map(String::valueOf).collect(Collectors.toList());
+            if (!Es6Util.checkWhetherUsePosition(id)) {
+                ids = Arrays.stream(StringUtils.split(id, ",")).map(String::valueOf).collect(Collectors.toList());
+            } else {
+                //compatible old version of generate doc' id
+                usePosition = true;
+                ids = Arrays.stream(StringUtils.split(id, ",")).map(Integer::valueOf).collect(Collectors.toList());
+            }
         }
         return this;
     }

elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/Es6Util.java

@@ -18,16 +18,20 @@
 
 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;
 
 import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
 
 import com.dtstack.flink.sql.util.DtStringUtil;
 
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * @author yinxi
@@ -63,5 +67,22 @@ 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;
+    }
 
 }