优化es6维表相关

Author: zoudaokoulife

core/src/main/java/com/dtstack/flink/sql/side/AllReqRow.java

@@ -21,9 +21,12 @@
 package com.dtstack.flink.sql.side;
 
 import com.dtstack.flink.sql.factory.DTThreadFactory;
+import org.apache.calcite.sql.JoinType;
 import org.apache.flink.api.common.functions.RichFlatMapFunction;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.runtime.types.CRow;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
 
 import java.sql.SQLException;
 import java.util.concurrent.Executors;
@@ -64,4 +67,13 @@ public void open(Configuration parameters) throws Exception {
         es.scheduleAtFixedRate(() -> reloadCache(), sideTableInfo.getCacheTimeout(), sideTableInfo.getCacheTimeout(), TimeUnit.MILLISECONDS);
     }
 
+    protected void sendOutputRow(CRow value, Object sideInput, Collector<CRow> out){
+        if(sideInput == null && sideInfo.getJoinType() != JoinType.LEFT){
+            return;
+        }
+
+        Row row = fillData(value.row(), sideInput);
+        out.collect(new CRow(row, value.change()));
+    }
+
 }

docs/elasticsearch6Side

@@ -70,7 +70,7 @@ create table sideTable(
     PRIMARY KEY(channel),
     PERIOD FOR SYSTEM_TIME
  )WITH(
-    type ='elasticsearch',
+    type ='elasticsearch6',
     address ='172.16.10.47:9500',
     cluster='es_47_menghan',
     estype ='type1',

elasticsearch6/elasticsearch6-side/elasticsearch6-all-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AllReqRow.java

@@ -80,11 +80,7 @@ public void flatMap(CRow value, Collector<CRow> out) throws Exception {
         for (Integer conValIndex : sideInfo.getEqualValIndex()) {
             Object equalObj = value.row().getField(conValIndex);
             if (equalObj == null) {
-                if (sideInfo.getJoinType() == JoinType.LEFT) {
-                    Row row = fillData(value.row(), null);
-                    out.collect(new CRow(row, value.change()));
-                }
-
+                sendOutputRow(value, null, out);
                 return;
             }
 
@@ -94,18 +90,12 @@ public void flatMap(CRow value, Collector<CRow> out) throws Exception {
         String key = buildKey(inputParams);
         List<Map<String, Object>> cacheList = cacheRef.get().get(key);
         if (CollectionUtils.isEmpty(cacheList)) {
-            if (sideInfo.getJoinType() == JoinType.LEFT) {
-                Row row = fillData(value.row(), null);
-                out.collect(new CRow(row, value.change()));
-            } else {
-                return;
-            }
-
+            sendOutputRow(value, null, out);
             return;
         }
 
         for (Map<String, Object> one : cacheList) {
-            out.collect(new CRow(fillData(value.row(), one), value.change()));
+            sendOutputRow(value, one, out);
         }
     }
 
@@ -155,7 +145,7 @@ private String buildKey(Map<String, Object> val, List<String> equalFieldList) {
     }
 
     @Override
-    protected void initCache() throws SQLException {
+    protected void initCache() {
         Map<String, List<Map<String, Object>>> newCache = Maps.newConcurrentMap();
         cacheRef.set(newCache);
         try {
@@ -165,6 +155,7 @@ protected void initCache() throws SQLException {
             loadData(newCache);
         } catch (Exception e) {
             LOG.error("", e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -176,6 +167,7 @@ protected void reloadCache() {
             loadData(newCache);
         } catch (Exception e) {
             LOG.error("", e);
+            throw new RuntimeException(e);
         }
 
         cacheRef.set(newCache);
@@ -210,6 +202,7 @@ private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws IO
 
         } catch (Exception e) {
             LOG.error("", e);
+            throw new RuntimeException(e);
         } finally {
 
             if (rhlClient != null) {
@@ -220,18 +213,18 @@ private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws IO
 
     // initialize searchSourceBuilder
     private SearchSourceBuilder initConfiguration(BoolQueryBuilder boolQueryBuilder){
-            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
-            if (boolQueryBuilder != null) {
-                searchSourceBuilder.query(boolQueryBuilder);
-            }
+        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+        if (boolQueryBuilder != null) {
+            searchSourceBuilder.query(boolQueryBuilder);
+        }
 
-            searchSourceBuilder.size(getFetchSize());
-            searchSourceBuilder.sort("_id", SortOrder.DESC);
+        searchSourceBuilder.size(getFetchSize());
+        searchSourceBuilder.sort("_id", SortOrder.DESC);
 
-            // fields included in the source data
-            String[] sideFieldNames = StringUtils.split(sideInfo.getSideSelectFields().trim(), ",");
-            searchSourceBuilder.fetchSource(sideFieldNames, null);
-            return searchSourceBuilder;
+        // fields included in the source data
+        String[] sideFieldNames = StringUtils.split(sideInfo.getSideSelectFields().trim(), ",");
+        searchSourceBuilder.fetchSource(sideFieldNames, null);
+        return searchSourceBuilder;
     }
 
 
@@ -247,6 +240,7 @@ private void searchData(SearchSourceBuilder searchSourceBuilder, Map<String, Lis
                     // set search mark
                     searchSourceBuilder.searchAfter(searchAfterParameter);
                 }
+
                 searchRequest.source(searchSourceBuilder);
                 searchResponse = rhlClient.search(searchRequest, RequestOptions.DEFAULT);
                 searchHits = searchResponse.getHits().getHits();
@@ -267,6 +261,7 @@ private void searchData(SearchSourceBuilder searchSourceBuilder, Map<String, Lis
     private void loadToCache(SearchHit[] searchHits, Map<String, List<Map<String, Object>>> tmpCache) {
         String[] sideFieldNames = StringUtils.split(sideInfo.getSideSelectFields().trim(), ",");
         String[] sideFieldTypes = sideInfo.getSideTableInfo().getFieldTypes();
+
         for (SearchHit searchHit : searchHits) {
             Map<String, Object> oneRow = Maps.newHashMap();
             for (String fieldName : sideFieldNames) {
@@ -275,6 +270,7 @@ private void loadToCache(SearchHit[] searchHits, Map<String, List<Map<String, Ob
                 object = SwitchUtil.getTarget(object, sideFieldTypes[fieldIndex]);
                 oneRow.put(fieldName.trim(), object);
             }
+
             String cacheKey = buildKey(oneRow, sideInfo.getEqualFieldList());
             List<Map<String, Object>> list = tmpCache.computeIfAbsent(cacheKey, key -> Lists.newArrayList());
             list.add(oneRow);