You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by dk...@apache.org on 2020/07/31 18:04:49 UTC
[hive] branch master updated: HIVE-23324: Parallelise compaction
directory cleaning process (Adesh Kumar Rao, reviewed by Peter Vary,
Beluga Behr, Sankar Hariappan, Denys Kuzmenko)
This is an automated email from the ASF dual-hosted git repository.
dkuzmenko pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new a206833 HIVE-23324: Parallelise compaction directory cleaning process (Adesh Kumar Rao, reviewed by Peter Vary, Beluga Behr, Sankar Hariappan, Denys Kuzmenko)
a206833 is described below
commit a206833639ebf69fa4c7d821f2ffca027c2f2ae5
Author: Adesh Kumar Rao <ad...@microsoft.com>
AuthorDate: Fri Jul 31 23:34:30 2020 +0530
HIVE-23324: Parallelise compaction directory cleaning process (Adesh Kumar Rao, reviewed by Peter Vary, Beluga Behr, Sankar Hariappan, Denys Kuzmenko)
Closes (#1275)
---
.../org/apache/hadoop/hive/conf/Constants.java | 3 +
.../java/org/apache/hadoop/hive/conf/HiveConf.java | 3 +
.../hadoop/hive/ql/txn/compactor/Cleaner.java | 77 ++++++++++++----------
.../hive/ql/txn/compactor/CompactorUtil.java | 53 +++++++++++++++
.../hadoop/hive/ql/txn/compactor/Initiator.java | 31 ++++-----
.../hadoop/hive/ql/txn/compactor/TestCleaner.java | 56 ++++++++++++++++
6 files changed, 170 insertions(+), 53 deletions(-)
diff --git a/common/src/java/org/apache/hadoop/hive/conf/Constants.java b/common/src/java/org/apache/hadoop/hive/conf/Constants.java
index a79be8d..c35ab15 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/Constants.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/Constants.java
@@ -83,4 +83,7 @@ public class Constants {
public static final String SCHEDULED_QUERY_EXECUTIONID = "scheduled.query.executionid";
public static final String SCHEDULED_QUERY_USER = "scheduled.query.user";
+ public static final String COMPACTOR_INTIATOR_THREAD_NAME_FORMAT = "Initiator-executor-thread-%d";
+ public static final String COMPACTOR_CLEANER_THREAD_NAME_FORMAT = "Cleaner-executor-thread-%d";
+
}
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index 928221d..0e539d3 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -3043,6 +3043,9 @@ public class HiveConf extends Configuration {
HIVE_COMPACTOR_CLEANER_RUN_INTERVAL("hive.compactor.cleaner.run.interval", "5000ms",
new TimeValidator(TimeUnit.MILLISECONDS), "Time between runs of the cleaner thread"),
+ HIVE_COMPACTOR_CLEANER_THREADS_NUM("hive.compactor.cleaner.threads.num", 1,
+ "Enables parallelization of the cleaning directories after compaction, that includes many file \n" +
+ "related checks and may be expensive"),
COMPACTOR_JOB_QUEUE("hive.compactor.job.queue", "", "Used to specify name of Hadoop queue to which\n" +
"Compaction jobs will be submitted. Set to empty string to let Hadoop choose the queue."),
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
index abde748..7ae3c01 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
@@ -51,9 +51,12 @@ import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import static org.apache.hadoop.hive.conf.Constants.COMPACTOR_CLEANER_THREAD_NAME_FORMAT;
import static org.apache.hadoop.hive.metastore.HiveMetaStore.HMSHandler.getMSForConf;
import static org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.getDefaultCatalog;
@@ -66,53 +69,59 @@ public class Cleaner extends MetaStoreCompactorThread {
private long cleanerCheckInterval = 0;
private ReplChangeManager replChangeManager;
+ private ExecutorService cleanerExecutor;
@Override
public void init(AtomicBoolean stop) throws Exception {
super.init(stop);
replChangeManager = ReplChangeManager.getInstance(conf);
+ cleanerCheckInterval = conf.getTimeVar(
+ HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_RUN_INTERVAL, TimeUnit.MILLISECONDS);
+ cleanerExecutor = CompactorUtil.createExecutorWithThreadFactory(
+ conf.getIntVar(HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_THREADS_NUM),
+ COMPACTOR_CLEANER_THREAD_NAME_FORMAT);
}
@Override
public void run() {
- if (cleanerCheckInterval == 0) {
- cleanerCheckInterval = conf.getTimeVar(
- HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_RUN_INTERVAL, TimeUnit.MILLISECONDS);
- }
-
- do {
- TxnStore.MutexAPI.LockHandle handle = null;
- long startedAt = -1;
- // Make sure nothing escapes this run method and kills the metastore at large,
- // so wrap it in a big catch Throwable statement.
- try {
- handle = txnHandler.getMutexAPI().acquireLock(TxnStore.MUTEX_KEY.Cleaner.name());
- startedAt = System.currentTimeMillis();
- long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
- clean(compactionInfo, minOpenTxnId);
- }
- } catch (Throwable t) {
- LOG.error("Caught an exception in the main loop of compactor cleaner, " +
- StringUtils.stringifyException(t));
- }
- finally {
- if (handle != null) {
- handle.releaseLocks();
- }
- }
- // Now, go back to bed until it's time to do this again
- long elapsedTime = System.currentTimeMillis() - startedAt;
- if (elapsedTime >= cleanerCheckInterval || stop.get()) {
- continue;
- } else {
+ try {
+ do {
+ TxnStore.MutexAPI.LockHandle handle = null;
+ long startedAt = -1;
+ // Make sure nothing escapes this run method and kills the metastore at large,
+ // so wrap it in a big catch Throwable statement.
try {
+ handle = txnHandler.getMutexAPI().acquireLock(TxnStore.MUTEX_KEY.Cleaner.name());
+ startedAt = System.currentTimeMillis();
+ long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
+ List<CompletableFuture> cleanerList = new ArrayList<>();
+ for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ }
+ CompletableFuture.allOf(cleanerList.toArray(new CompletableFuture[0])).join();
+ } catch (Throwable t) {
+ LOG.error("Caught an exception in the main loop of compactor cleaner, " +
+ StringUtils.stringifyException(t));
+ } finally {
+ if (handle != null) {
+ handle.releaseLocks();
+ }
+ }
+ // Now, go back to bed until it's time to do this again
+ long elapsedTime = System.currentTimeMillis() - startedAt;
+ if (elapsedTime < cleanerCheckInterval && !stop.get()) {
Thread.sleep(cleanerCheckInterval - elapsedTime);
- } catch (InterruptedException ie) {
- // What can I do about it?
}
+ } while (!stop.get());
+ } catch (InterruptedException ie) {
+ LOG.error("Compactor cleaner thread interrupted, exiting " +
+ StringUtils.stringifyException(ie));
+ } finally {
+ if (cleanerExecutor != null) {
+ this.cleanerExecutor.shutdownNow();
}
- } while (!stop.get());
+ }
}
private void clean(CompactionInfo ci, long minOpenTxnGLB) throws MetaException {
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorUtil.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorUtil.java
new file mode 100644
index 0000000..85c1bf6
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorUtil.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.txn.compactor;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.hadoop.hive.conf.HiveConf;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+
+public class CompactorUtil {
+ public interface ThrowingRunnable<E extends Exception> {
+ void run() throws E;
+
+ static Runnable unchecked(ThrowingRunnable<?> r) {
+ return () -> {
+ try {
+ r.run();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ };
+ }
+ }
+
+ public static ThreadFactory createThreadFactory(String threadNameFormat) {
+ return new ThreadFactoryBuilder()
+ .setPriority(Thread.currentThread().getPriority())
+ .setDaemon(Thread.currentThread().isDaemon())
+ .setNameFormat(threadNameFormat)
+ .build();
+ }
+
+ public static ExecutorService createExecutorWithThreadFactory(int threadCount, String threadNameFormat) {
+ return Executors.newFixedThreadPool(threadCount, createThreadFactory(threadNameFormat));
+ }
+}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java
index d35e229..f23d6b7 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java
@@ -62,13 +62,14 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.Executor;
-import java.util.concurrent.Executors;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
+import static org.apache.hadoop.hive.conf.Constants.COMPACTOR_INTIATOR_THREAD_NAME_FORMAT;
+
/**
* A class to initiate compactions. This will run in a separate thread.
* It's critical that there exactly 1 of these in a given warehouse.
@@ -78,10 +79,10 @@ public class Initiator extends MetaStoreCompactorThread {
static final private Logger LOG = LoggerFactory.getLogger(CLASS_NAME);
static final private String COMPACTORTHRESHOLD_PREFIX = "compactorthreshold.";
- private Executor compactionExecutor;
private long checkInterval;
private long prevStart = -1;
+ private ExecutorService compactionExecutor;
@Override
public void run() {
@@ -149,7 +150,7 @@ public class Initiator extends MetaStoreCompactorThread {
String runAs = resolveUserToRunAs(tblNameOwners, t, p);
/* checkForCompaction includes many file metadata checks and may be expensive.
* Therefore, using a thread pool here and running checkForCompactions in parallel */
- compactionList.add(CompletableFuture.runAsync(ThrowingRunnable.unchecked(() ->
+ compactionList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
scheduleCompactionIfRequired(ci, t, p, runAs)), compactionExecutor));
} catch (Throwable t) {
LOG.error("Caught exception while trying to determine if we should compact {}. " +
@@ -182,6 +183,10 @@ public class Initiator extends MetaStoreCompactorThread {
} catch (Throwable t) {
LOG.error("Caught an exception in the main loop of compactor initiator, exiting " +
StringUtils.stringifyException(t));
+ } finally {
+ if (compactionExecutor != null) {
+ this.compactionExecutor.shutdownNow();
+ }
}
}
@@ -222,25 +227,13 @@ public class Initiator extends MetaStoreCompactorThread {
return cache.get(fullTableName);
}
- private interface ThrowingRunnable<E extends Exception> {
- void run() throws E;
-
- static Runnable unchecked(ThrowingRunnable<?> r) {
- return () -> {
- try {
- r.run();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- };
- }
- }
-
@Override
public void init(AtomicBoolean stop) throws Exception {
super.init(stop);
checkInterval = conf.getTimeVar(HiveConf.ConfVars.HIVE_COMPACTOR_CHECK_INTERVAL, TimeUnit.MILLISECONDS);
- compactionExecutor = Executors.newFixedThreadPool(conf.getIntVar(HiveConf.ConfVars.HIVE_COMPACTOR_REQUEST_QUEUE));
+ compactionExecutor = CompactorUtil.createExecutorWithThreadFactory(
+ conf.getIntVar(HiveConf.ConfVars.HIVE_COMPACTOR_REQUEST_QUEUE),
+ COMPACTOR_INTIATOR_THREAD_NAME_FORMAT);
}
private void recoverFailedCompactions(boolean remoteOnly) throws MetaException {
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java b/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java
index 46c79d6..a806c16 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hive.ql.txn.compactor;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.CompactionRequest;
import org.apache.hadoop.hive.metastore.api.CompactionType;
import org.apache.hadoop.hive.metastore.api.Partition;
@@ -30,6 +31,7 @@ import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -274,6 +276,60 @@ public class TestCleaner extends CompactorTest {
ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
Assert.assertEquals(0, rsp.getCompactsSize());
}
+
+ @Test
+ public void processCompactionCandidatesInParallel() throws Exception {
+ Table t = newTable("default", "camipc", true);
+ List<Partition> partitions = new ArrayList<>();
+ Partition p = null;
+ for(int i = 0; i < 10; i++) {
+ p = newPartition(t, "today" + i);
+
+ addBaseFile(t, p, 20L, 20);
+ addDeltaFile(t, p, 21L, 22L, 2);
+ addDeltaFile(t, p, 23L, 24L, 2);
+ addDeltaFile(t, p, 21L, 24L, 4);
+ partitions.add(p);
+ }
+
+ burnThroughTransactions("default", "camipc", 25);
+ for(int i = 0; i < 10; i++) {
+ CompactionRequest rqst = new CompactionRequest("default", "camipc", CompactionType.MINOR);
+ rqst.setPartitionname("ds=today"+i);
+ txnHandler.compact(rqst);
+ CompactionInfo ci = txnHandler.findNextToCompact("fred");
+ ci.runAs = System.getProperty("user.name");
+ txnHandler.updateCompactorState(ci, openTxn());
+ txnHandler.markCompacted(ci);
+ }
+
+ conf.setIntVar(HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_THREADS_NUM, 3);
+ startCleaner();
+
+ // Check there are no compactions requests left.
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(10, rsp.getCompactsSize());
+ Assert.assertTrue(TxnStore.SUCCEEDED_RESPONSE.equals(rsp.getCompacts().get(0).getState()));
+
+ // Check that the files are removed
+ for (Partition pa : partitions) {
+ List<Path> paths = getDirectories(conf, t, pa);
+ Assert.assertEquals(2, paths.size());
+ boolean sawBase = false, sawDelta = false;
+ for (Path path : paths) {
+ if (path.getName().equals("base_20")) {
+ sawBase = true;
+ } else if (path.getName().equals(makeDeltaDirNameCompacted(21, 24))) {
+ sawDelta = true;
+ } else {
+ Assert.fail("Unexpected file " + path.getName());
+ }
+ }
+ Assert.assertTrue(sawBase);
+ Assert.assertTrue(sawDelta);
+ }
+ }
+
@Override
boolean useHive130DeltaDirName() {
return false;