id allow as null

Author: todd5167

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

@@ -35,6 +35,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * Reason:
@@ -47,6 +48,8 @@ public class CustomerSinkFunc implements ElasticsearchSinkFunction<Tuple2> {
 
     private final Logger logger = LoggerFactory.getLogger(CustomerSinkFunc.class);
 
+    private static final String ID_VALUE_SPLIT = "_";
+
     private String index;
 
     private String type;
@@ -93,27 +96,29 @@ public void setOutRecords(Counter outRecords) {
     }
 
     private IndexRequest createIndexRequest(Row element) {
-
-        List<String> idFieldList = new ArrayList<>();
-        for(int index : idFieldIndexList){
-            if(index >= element.getArity()){
-                continue;
-            }
-
-            idFieldList.add(element.getField(index).toString());
-        }
+        // index start at 1,
+        String 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 id = StringUtils.join(idFieldList, sp);
+        if (StringUtils.isEmpty(idFieldStr)) {
+            return Requests.indexRequest()
+                    .index(index)
+                    .type(type)
+                    .source(dataMap);
+        }
+
         return Requests.indexRequest()
                 .index(index)
                 .type(type)
-                .id(id)
+                .id(idFieldStr)
                 .source(dataMap);
     }
 }

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

@@ -132,12 +132,13 @@ public boolean check() {
         Preconditions.checkNotNull(address, "elasticsearch type of address is required");
         Preconditions.checkNotNull(index, "elasticsearch type of index is required");
         Preconditions.checkNotNull(esType, "elasticsearch type of type is required");
-        Preconditions.checkNotNull(id, "elasticsearch type of id is required");
         Preconditions.checkNotNull(clusterName, "elasticsearch type of clusterName is required");
 
-        Arrays.stream(StringUtils.split(id, ",")).forEach(number -> {
-            Preconditions.checkArgument(NumberUtils.isNumber(number), "id must be a numeric type");
-        });
+        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");

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

@@ -127,14 +127,6 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
         }
     }
 
-    public void setParallelism(int parallelism) {
-        this.parallelism = parallelism;
-    }
-
-    public void setBulkFlushMaxActions(int bulkFlushMaxActions) {
-        this.bulkFlushMaxActions = bulkFlushMaxActions;
-    }
-
     @Override
     public ElasticsearchSink genStreamSink(TargetTableInfo targetTableInfo) {
         esTableInfo = (ElasticsearchTableInfo) targetTableInfo;
@@ -143,8 +135,8 @@ public ElasticsearchSink genStreamSink(TargetTableInfo targetTableInfo) {
         type = esTableInfo.getEsType();
         columnTypes = esTableInfo.getFieldTypes();
         esAddressList = Arrays.asList(esTableInfo.getAddress().split(","));
-
         String id = esTableInfo.getId();
+
         if (!StringUtils.isEmpty(id)) {
             idIndexList = Arrays.stream(StringUtils.split(id, ",")).map(Integer::valueOf).collect(Collectors.toList());
         }

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

@@ -128,12 +128,13 @@ public boolean check() {
         Preconditions.checkNotNull(address, "elasticsearch6 type of address is required");
         Preconditions.checkNotNull(index, "elasticsearch6 type of index is required");
         Preconditions.checkNotNull(esType, "elasticsearch6 type of type is required");
-        Preconditions.checkNotNull(id, "elasticsearch6 type of id is required");
         Preconditions.checkNotNull(clusterName, "elasticsearch6 type of clusterName is required");
 
-        Arrays.stream(StringUtils.split(id, ",")).forEach(number ->{
-            Preconditions.checkArgument(NumberUtils.isNumber(number),"id must be a numeric type");
-        });
+        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, "elasticsearch6 type of userName is required");