[kylin] branch master updated: KYLIN-4573 Add option to indicate whether to close file for every append for Hive Producer

[xx...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

xxyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kylin.git


The following commit(s) were added to refs/heads/master by this push:
     new 616e066  KYLIN-4573 Add option to indicate whether to close file for every append for Hive Producer
616e066 is described below

commit 616e06675278a6857f3cbb353a4f9c2243eeccc1
Author: Zhong, Yanghong <nj...@apache.org>
AuthorDate: Mon Jun 15 16:41:36 2020 +0800

    KYLIN-4573 Add option to indicate whether to close file for every append for Hive Producer
---
 .../kylin/metrics/lib/impl/hive/HiveProducer.java      | 18 ++++++++++++++++--
 server/src/main/resources/kylinMetrics.xml             |  1 +
 2 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/metrics-reporter-hive/src/main/java/org/apache/kylin/metrics/lib/impl/hive/HiveProducer.java b/metrics-reporter-hive/src/main/java/org/apache/kylin/metrics/lib/impl/hive/HiveProducer.java
index 1ab3f89..8bc7a43 100644
--- a/metrics-reporter-hive/src/main/java/org/apache/kylin/metrics/lib/impl/hive/HiveProducer.java
+++ b/metrics-reporter-hive/src/main/java/org/apache/kylin/metrics/lib/impl/hive/HiveProducer.java
@@ -48,6 +48,7 @@ import org.slf4j.LoggerFactory;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -72,6 +73,10 @@ public class HiveProducer {
      */
     private final boolean supportAppend;
 
+    private final boolean closeFileEveryAppend;
+
+    private final Map<String, String> kylinSpecifiedConfig = new HashMap<>();
+
     public HiveProducer(String metricType, Properties props) throws Exception {
         this(metricType, props, new HiveConf());
     }
@@ -80,7 +85,13 @@ public class HiveProducer {
         this.metricType = metricType;
         hiveConf = hiveConfig;
         for (Map.Entry<Object, Object> e : props.entrySet()) {
-            hiveConf.set(e.getKey().toString(), e.getValue().toString());
+            String key = e.getKey().toString();
+            String value = e.getValue().toString();
+            if (key.startsWith("kylin.")) {
+                kylinSpecifiedConfig.put(key, value);
+            } else {
+                hiveConf.set(key, value);
+            }
         }
 
         fs = FileSystem.get(hiveConf);
@@ -119,6 +130,9 @@ public class HiveProducer {
         String fsUri = fs.getUri().toString();
         supportAppend = fsUri.startsWith("hdfs") ; // Only HDFS is appendable
         logger.info("For {}, supportAppend was set to {}", fsUri, supportAppend);
+
+        closeFileEveryAppend = !supportAppend
+                || Boolean.parseBoolean(kylinSpecifiedConfig.get("kylin.hive.producer.close-file-every-append"));
     }
 
     public void close() {
@@ -270,7 +284,7 @@ public class HiveProducer {
                     + " due to ", e);
             closeFout();
         }
-        if (!supportAppend) {
+        if (closeFileEveryAppend) {
             closeFout();
         }
     }
diff --git a/server/src/main/resources/kylinMetrics.xml b/server/src/main/resources/kylinMetrics.xml
index 843fb91..85c879f 100644
--- a/server/src/main/resources/kylinMetrics.xml
+++ b/server/src/main/resources/kylinMetrics.xml
@@ -73,6 +73,7 @@
                                           value="org.apache.kylin.metrics.lib.impl.hive.HiveReservoirReporter"/>
                                 <property name="second">
                                     <props>
+                                        <prop key="kylin.hive.producer.close-file-every-append">true</prop>
                                     </props>
                                 </property>
                             </bean>