[hotfix-34999][elasticsearch6][sink]optimize es6's pom and adjusted generation strategy of doc's id to field name.

Author: HiLany

elasticsearch6/elasticsearch6-side/elasticsearch6-all-side/pom.xml

@@ -15,7 +15,6 @@
 
     <properties>
         <sql.side.elasticsearch6.core.version>1.0-SNAPSHOT</sql.side.elasticsearch6.core.version>
-        <elasticsearch.version>6.8.6</elasticsearch.version>
     </properties>
 
 

elasticsearch6/elasticsearch6-side/elasticsearch6-async-side/pom.xml

@@ -15,7 +15,6 @@
 
     <properties>
         <sql.side.elasticsearch6.core.version>1.0-SNAPSHOT</sql.side.elasticsearch6.core.version>
-        <elasticsearch.version>6.8.6</elasticsearch.version>
     </properties>
 
 

elasticsearch6/elasticsearch6-side/elasticsearch6-side-core/pom.xml

@@ -15,10 +15,6 @@
     <version>1.0-SNAPSHOT</version>
     <packaging>jar</packaging>
 
-    <properties>
-        <elasticsearch.version>6.8.6</elasticsearch.version>
-    </properties>
-
     <dependencies>
         <dependency>
             <groupId>org.elasticsearch.client</groupId>

elasticsearch6/elasticsearch6-sink/pom.xml

@@ -12,10 +12,6 @@
     <artifactId>sql.sink.elasticsearch6</artifactId>
     <name>elasticsearch6-sink</name>
 
-    <properties>
-        <elasticsearch.version>6.8.6</elasticsearch.version>
-    </properties>
-
     <dependencies>
         <dependency>
             <groupId>org.apache.flink</groupId>

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

@@ -53,7 +53,7 @@ public class CustomerSinkFunc implements ElasticsearchSinkFunction<Tuple2> {
 
     private String type;
 
-    private List<Integer> idFieldIndexList;
+    private List<String> idFiledNames;
 
     private List<String> fieldNames;
 
@@ -63,12 +63,12 @@ public class CustomerSinkFunc implements ElasticsearchSinkFunction<Tuple2> {
 
     private transient Counter outDirtyRecords;
 
-    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<String> idFiledNames) {
         this.index = index;
         this.type = type;
         this.fieldNames = fieldNames;
         this.fieldTypes = fieldTypes;
-        this.idFieldIndexList = idFieldIndexes;
+        this.idFiledNames = idFiledNames;
     }
 
     @Override
@@ -99,14 +99,6 @@ public void setOutDirtyRecords(Counter outDirtyRecords) {
     }
 
     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 = Es6Util.rowToJsonMap(element, fieldNames, fieldTypes);
         int length = Math.min(element.getArity(), fieldNames.size());
@@ -118,6 +110,12 @@ private IndexRequest createIndexRequest(Row element) {
             dataMap.put(fieldNames.get(i), element.getField(i));
         }
 
+        String idFieldStr = "";
+        if (null != idFiledNames) {
+            idFieldStr = idFiledNames.stream()
+                    .map(idFiledName -> dataMap.get(idFiledName).toString())
+                    .collect(Collectors.joining(ID_VALUE_SPLIT));
+        }
 
         if (StringUtils.isEmpty(idFieldStr)) {
             return Requests.indexRequest()

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

@@ -19,6 +19,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;
@@ -61,7 +62,7 @@ public class ElasticsearchSink implements RetractStreamTableSink<Row>, IStreamSi
 
     private String type = "";
 
-    private List<Integer> idIndexList;
+    private List<String> idFiledNames;
 
     protected String[] fieldNames;
 
@@ -105,6 +106,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);
+
         Map<String, String> userConfig = Maps.newHashMap();
         userConfig.put("cluster.name", clusterName);
         // This instructs the sink to emit after every element, otherwise they would be buffered
@@ -118,7 +125,7 @@ private RichSinkFunction createEsSinkFunction() {
                     return new HttpHost(host.trim(), port, ES_DEFAULT_SCHEMA);
                 }).collect(Collectors.toList());
 
-        CustomerSinkFunc customerSinkFunc = new CustomerSinkFunc(index, type, Arrays.asList(fieldNames), Arrays.asList(columnTypes), idIndexList);
+
         return new MetricElasticsearch6Sink(userConfig, transports, customerSinkFunc, esTableInfo);
     }
 
@@ -150,7 +157,7 @@ public ElasticsearchSink genStreamSink(AbstractTargetTableInfo targetTableInfo)
         parallelism = Objects.isNull(esTableInfo.getParallelism()) ? parallelism : esTableInfo.getParallelism();
 
         if (!StringUtils.isEmpty(id)) {
-            idIndexList = Arrays.stream(StringUtils.split(id, ",")).map(Integer::valueOf).collect(Collectors.toList());
+            idFiledNames = Arrays.stream(StringUtils.split(id, ",")).map(String::valueOf).collect(Collectors.toList());
         }
         return this;
     }

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

@@ -130,12 +130,6 @@ public boolean check() {
         Preconditions.checkNotNull(esType, "elasticsearch6 type of type is required");
         Preconditions.checkNotNull(clusterName, "elasticsearch6 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, "elasticsearch6 type of userName is required");
             Preconditions.checkNotNull(password, "elasticsearch6 type of password is required");

elasticsearch6/pom.xml

@@ -11,6 +11,12 @@
 
     <artifactId>sql.elasticsearch6</artifactId>
     <packaging>pom</packaging>
+
+
+    <properties>
+        <elasticsearch.version>6.8.6</elasticsearch.version>
+    </properties>
+
     <modules>
         <module>elasticsearch6-side</module>
         <module>elasticsearch6-sink</module>