You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by ni...@apache.org on 2019/07/11 22:52:22 UTC
[kylin] 01/02: KYLIN-4026: Avoid too many file append operations in
HiveProducer of hive metrics reporter
This is an automated email from the ASF dual-hosted git repository.
nic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kylin.git
commit 46d3f09ff777fdafc69ed43f169b5cbf17fa4569
Author: Liu Shaohui <li...@xiaomi.com>
AuthorDate: Fri May 31 19:59:59 2019 +0800
KYLIN-4026: Avoid too many file append operations in HiveProducer of hive metrics reporter
---
.../kylin/metrics/lib/impl/BlockingReservoir.java | 11 +--
metrics-reporter-hive/pom.xml | 5 ++
.../kylin/metrics/lib/impl/hive/HiveProducer.java | 84 ++++++++++++++++------
.../lib/impl/hive/HiveReservoirReporter.java | 43 +++++++++--
4 files changed, 113 insertions(+), 30 deletions(-)
diff --git a/core-metrics/src/main/java/org/apache/kylin/metrics/lib/impl/BlockingReservoir.java b/core-metrics/src/main/java/org/apache/kylin/metrics/lib/impl/BlockingReservoir.java
index 3301867..cbf1162 100644
--- a/core-metrics/src/main/java/org/apache/kylin/metrics/lib/impl/BlockingReservoir.java
+++ b/core-metrics/src/main/java/org/apache/kylin/metrics/lib/impl/BlockingReservoir.java
@@ -33,6 +33,7 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder;
public class BlockingReservoir extends AbstractActiveReservoir {
private static final Logger logger = LoggerFactory.getLogger(BlockingReservoir.class);
+ private static final int MAX_QUEUE_SIZE = 50000;
private final BlockingQueue<Record> recordsQueue;
private final Thread scheduledReporter;
@@ -42,7 +43,7 @@ public class BlockingReservoir extends AbstractActiveReservoir {
private List<Record> records;
public BlockingReservoir() {
- this(1, 100);
+ this(100, 500);
}
public BlockingReservoir(int minReportSize, int maxReportSize) {
@@ -54,7 +55,7 @@ public class BlockingReservoir extends AbstractActiveReservoir {
this.MIN_REPORT_SIZE = minReportSize;
this.MAX_REPORT_TIME = MAX_REPORT_TIME * 60 * 1000L;
- this.recordsQueue = new LinkedBlockingQueue<>();
+ this.recordsQueue = new LinkedBlockingQueue<>(MAX_QUEUE_SIZE);
this.listeners = Lists.newArrayList();
this.records = Lists.newArrayListWithExpectedSize(MAX_REPORT_SIZE);
@@ -72,6 +73,8 @@ public class BlockingReservoir extends AbstractActiveReservoir {
recordsQueue.put(record);
} catch (InterruptedException e) {
logger.warn("Thread is interrupted during putting value to blocking queue. \n" + e.toString());
+ } catch (IllegalArgumentException e) {
+ logger.warn("The record queue may be full");
}
}
@@ -93,7 +96,7 @@ public class BlockingReservoir extends AbstractActiveReservoir {
if (!notifyListenerOfUpdatedRecord(listener, records)) {
ifSucceed = false;
logger.warn("It fails to notify listener " + listener.toString() + " of updated records "
- + records.toString());
+ + records.size());
}
}
if (!ifSucceed) {
@@ -106,7 +109,7 @@ public class BlockingReservoir extends AbstractActiveReservoir {
}
private boolean notifyListenerHAOfUpdatedRecord(List<Record> records) {
- logger.info("The HA listener " + listenerHA.toString() + " for updated records " + records.toString()
+ logger.info("The HA listener " + listenerHA.toString() + " for updated records " + records.size()
+ " will be started");
if (!notifyListenerOfUpdatedRecord(listenerHA, records)) {
logger.error("The HA listener also fails!!!");
diff --git a/metrics-reporter-hive/pom.xml b/metrics-reporter-hive/pom.xml
index 321ad77..0c79050 100644
--- a/metrics-reporter-hive/pom.xml
+++ b/metrics-reporter-hive/pom.xml
@@ -48,5 +48,10 @@
<artifactId>hadoop-common</artifactId>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <scope>provided</scope>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
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 04a1f20..2b09eba 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
@@ -29,6 +29,7 @@ import java.util.Properties;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hive.cli.CliSessionState;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
@@ -57,22 +58,30 @@ public class HiveProducer {
private static final int CACHE_MAX_SIZE = 10;
+ private String metricType;
private final HiveConf hiveConf;
- private final FileSystem hdfs;
+ private final FileSystem fs;
private final LoadingCache<Pair<String, String>, Pair<String, List<FieldSchema>>> tableFieldSchemaCache;
- private final String CONTENT_FILE_NAME;
+ private final String CONTENT_FILE_PREFIX;
- public HiveProducer(Properties props) throws Exception {
- this(props, new HiveConf());
+ private String prePartitionPath;
+ private Path curPartitionContentPath;
+ private int id = 0;
+ private FSDataOutputStream fout;
+
+
+ public HiveProducer(String metricType, Properties props) throws Exception {
+ this(metricType, props, new HiveConf());
}
- HiveProducer(Properties props, HiveConf hiveConfig) throws Exception {
+ HiveProducer(String metricType, Properties props, HiveConf hiveConfig) throws Exception {
+ this.metricType = metricType;
hiveConf = hiveConfig;
for (Map.Entry<Object, Object> e : props.entrySet()) {
hiveConf.set(e.getKey().toString(), e.getValue().toString());
}
- hdfs = FileSystem.get(hiveConf);
+ fs = FileSystem.get(hiveConf);
tableFieldSchemaCache = CacheBuilder.newBuilder().removalListener(new RemovalListener<Pair<String, String>, Pair<String, List<FieldSchema>>>() {
@Override
@@ -96,7 +105,7 @@ public class HiveProducer {
} catch (UnknownHostException e) {
hostName = "UNKNOWN";
}
- CONTENT_FILE_NAME = hostName + "-part-0000";
+ CONTENT_FILE_PREFIX = hostName + "-" + System.currentTimeMillis() + "-part-";
}
public void close() {
@@ -134,7 +143,7 @@ public class HiveProducer {
sb.append(e.getValue());
}
Path partitionPath = new Path(sb.toString());
- if (!hdfs.exists(partitionPath)) {
+ if (!fs.exists(partitionPath)) {
StringBuilder hql = new StringBuilder();
hql.append("ALTER TABLE ");
hql.append(recordKey.database() + "." + recordKey.table());
@@ -155,27 +164,62 @@ public class HiveProducer {
driver.run(hql.toString());
driver.close();
}
- Path partitionContentPath = new Path(partitionPath, CONTENT_FILE_NAME);
- if (!hdfs.exists(partitionContentPath)) {
+
+ if (fout == null || prePartitionPath == null || prePartitionPath.compareTo(partitionPath.toString()) != 0) {
+ if (fout != null) {
+ closeFout();
+ }
+
+ Path partitionContentPath = new Path(partitionPath, CONTENT_FILE_PREFIX + String.format(Locale.ROOT, "%04d", id));
+ logger.info("Try to use new partition content path: " + partitionContentPath.toString() + " for metric: " + metricType);
+ if (!fs.exists(partitionContentPath)) {
int nRetry = 0;
- while (!hdfs.createNewFile(partitionContentPath) && nRetry++ < 5) {
- if (hdfs.exists(partitionContentPath)) {
- break;
- }
- Thread.sleep(500L * nRetry);
+ while (!fs.createNewFile(partitionContentPath) && nRetry++ < 5) {
+ if (fs.exists(partitionContentPath)) {
+ break;
+ }
+ Thread.sleep(500L * nRetry);
}
- if (!hdfs.exists(partitionContentPath)) {
- throw new RuntimeException("Fail to create HDFS file: " + partitionContentPath + " after " + nRetry + " retries");
+ if (!fs.exists(partitionContentPath)) {
+ throw new RuntimeException("Fail to create HDFS file: " + partitionContentPath + " after " + nRetry + " retries");
}
+ }
+ fout = fs.append(partitionContentPath);
+ prePartitionPath = partitionPath.toString();
+ curPartitionContentPath = partitionContentPath;
+ id = (id + 1) % 10;
}
- try (FSDataOutputStream fout = hdfs.append(partitionContentPath)) {
+
+ try {
+ int count = 0;
for (HiveProducerRecord elem : recordItr) {
fout.writeBytes(elem.valueToString() + "\n");
+ count++;
}
+ logger.info("Success to write " + count + " metrics(" + metricType + ") to file " + curPartitionContentPath.toString());
} catch (IOException e) {
- System.out.println("Fails to write metrics to file " + partitionContentPath.toString() + " due to " + e);
- logger.error("Fails to write metrics to file " + partitionContentPath.toString() + " due to " + e);
+ logger.error("Fails to write metrics(" + metricType + ") to file " + curPartitionContentPath.toString() + " due to ", e);
+ closeFout();
+ }
+ }
+
+ private void closeFout() {
+ if (fout != null) {
+ try {
+ fout.close();
+ } catch (Exception e) {
+ logger.error("Close the path: " + curPartitionContentPath + " failed", e);
+ if (fs instanceof DistributedFileSystem) {
+ DistributedFileSystem hdfs = (DistributedFileSystem) fs;
+ try {
+ boolean recovered = hdfs.recoverLease(curPartitionContentPath);
+ } catch (Exception e1) {
+ logger.error("Recover lease for path: " + curPartitionContentPath + " failed", e1);
+ }
+ }
}
+ }
+ fout = null;
}
private HiveProducerRecord convertTo(Record record) throws Exception {
diff --git a/metrics-reporter-hive/src/main/java/org/apache/kylin/metrics/lib/impl/hive/HiveReservoirReporter.java b/metrics-reporter-hive/src/main/java/org/apache/kylin/metrics/lib/impl/hive/HiveReservoirReporter.java
index 5af2bf9..4008005 100644
--- a/metrics-reporter-hive/src/main/java/org/apache/kylin/metrics/lib/impl/hive/HiveReservoirReporter.java
+++ b/metrics-reporter-hive/src/main/java/org/apache/kylin/metrics/lib/impl/hive/HiveReservoirReporter.java
@@ -18,7 +18,10 @@
package org.apache.kylin.metrics.lib.impl.hive;
+import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
import org.apache.kylin.metrics.lib.ActiveReservoir;
@@ -105,16 +108,40 @@ public class HiveReservoirReporter extends ActiveReservoirReporter {
}
private class HiveReservoirListener implements ActiveReservoirListener {
-
- HiveProducer producer;
-
+ private Properties props;
+ private Map<String, HiveProducer> producerMap = new HashMap<String, HiveProducer>();
private HiveReservoirListener(Properties props) throws Exception {
- producer = new HiveProducer(props);
+ this.props = props;
}
+ private synchronized HiveProducer getProducer(String metricType)throws Exception {
+ HiveProducer producer = producerMap.get(metricType);
+ if (producer == null) {
+ producer = new HiveProducer(metricType, props);
+ producerMap.put(metricType, producer);
+ }
+ return producer;
+ }
public boolean onRecordUpdate(final List<Record> records) {
+ if (records.size() == 0) {
+ return true;
+ }
+ logger.info("Try to write " + records.size() + " records");
try {
- producer.send(records);
+ Map<String, List<Record>> queues = new HashMap<>();
+ for (Record record: records) {
+ List<Record> recordQueues = queues.get(record.getType());
+ if (recordQueues == null) {
+ recordQueues = new ArrayList<>();
+ queues.put(record.getType(), recordQueues);
+ }
+ recordQueues.add(record);
+ }
+ for (Map.Entry<String, List<Record>> entry: queues.entrySet()) {
+ HiveProducer producer = getProducer(entry.getKey());
+ producer.send(entry.getValue());
+ }
+ queues.clear();
} catch (Exception e) {
logger.error(e.getMessage(), e);
return false;
@@ -124,6 +151,7 @@ public class HiveReservoirReporter extends ActiveReservoirReporter {
public boolean onRecordUpdate(final Record record) {
try {
+ HiveProducer producer = getProducer(record.getType());
producer.send(record);
} catch (Exception e) {
logger.error(e.getMessage(), e);
@@ -133,7 +161,10 @@ public class HiveReservoirReporter extends ActiveReservoirReporter {
}
public void close() {
- producer.close();
+ for(HiveProducer producer: producerMap.values()) {
+ producer.close();
+ }
+ producerMap.clear();
}
}
}