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

Author: HiLany

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

@@ -53,20 +53,24 @@ 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;
 
     private List<String> fieldTypes;
 
     public transient Counter outRecords;
 
-    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
@@ -100,10 +104,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)) {

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

@@ -47,6 +47,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 /**
  * table output elastic5plugin
@@ -69,7 +70,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;
 
@@ -114,10 +118,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), "elasticsearch5 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 = new HashMap<>();
         userConfig.put("cluster.name", clusterName);
@@ -182,15 +188,21 @@ public ElasticsearchSink genStreamSink(AbstractTargetTableInfo targetTableInfo)
         type = elasticsearchTableInfo.getEsType();
         String id = elasticsearchTableInfo.getId();
         String[] idField = StringUtils.split(id, ",");
-        idFiledNames = new ArrayList<>();
         registerTableName = elasticsearchTableInfo.getName();
         parallelism = Objects.isNull(elasticsearchTableInfo.getParallelism()) ?
                 parallelism : elasticsearchTableInfo.getParallelism();
 
         for(int i = 0; i < idField.length; ++i) {
-            idFiledNames.add(String.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;
+    }
+
 
 }