hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From pv...@apache.org
Subject hive git commit: HIVE-17362: The MAX_PREWARM_TIME should be configurable on HoS (Peter Vary, reviewed by Xuefu Zhang)
Date Wed, 23 Aug 2017 09:29:22 GMT
Repository: hive
Updated Branches:
  refs/heads/master 7fa94d9eb -> a2c4aaede


HIVE-17362: The MAX_PREWARM_TIME should be configurable on HoS (Peter Vary, reviewed by Xuefu
Zhang)


Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/a2c4aaed
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/a2c4aaed
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/a2c4aaed

Branch: refs/heads/master
Commit: a2c4aaede434c17fe017d588c8f3be174e47bf37
Parents: 7fa94d9
Author: Peter Vary <pvary@cloudera.com>
Authored: Wed Aug 23 11:28:35 2017 +0200
Committer: Peter Vary <pvary@cloudera.com>
Committed: Wed Aug 23 11:28:35 2017 +0200

----------------------------------------------------------------------
 .../org/apache/hadoop/hive/conf/HiveConf.java   |  4 ++-
 data/conf/spark/standalone/hive-site.xml        | 15 ++++++++++
 data/conf/spark/yarn-client/hive-site.xml       | 10 +++++++
 .../org/apache/hadoop/hive/ql/QTestUtil.java    | 31 ++------------------
 .../ql/exec/spark/RemoteHiveSparkClient.java    | 14 +++++----
 5 files changed, 38 insertions(+), 36 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/a2c4aaed/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
----------------------------------------------------------------------
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 37eb8fe..dca6e81 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -2844,7 +2844,9 @@ public class HiveConf extends Configuration {
         "Whether to generate consistent split locations when generating splits in the AM"),
     HIVE_PREWARM_ENABLED("hive.prewarm.enabled", false, "Enables container prewarm for Tez/Spark
(Hadoop 2 only)"),
     HIVE_PREWARM_NUM_CONTAINERS("hive.prewarm.numcontainers", 10, "Controls the number of
containers to prewarm for Tez/Spark (Hadoop 2 only)"),
-
+    HIVE_PREWARM_SPARK_TIMEOUT("hive.prewarm.spark.timeout", "5000ms",
+         new TimeValidator(TimeUnit.MILLISECONDS),
+         "Time to wait to finish prewarming spark executors"),
     HIVESTAGEIDREARRANGE("hive.stageid.rearrange", "none", new StringSet("none", "idonly",
"traverse", "execution"), ""),
     HIVEEXPLAINDEPENDENCYAPPENDTASKTYPES("hive.explain.dependency.append.tasktype", false,
""),
 

http://git-wip-us.apache.org/repos/asf/hive/blob/a2c4aaed/data/conf/spark/standalone/hive-site.xml
----------------------------------------------------------------------
diff --git a/data/conf/spark/standalone/hive-site.xml b/data/conf/spark/standalone/hive-site.xml
index a3b52c7..90112e9 100644
--- a/data/conf/spark/standalone/hive-site.xml
+++ b/data/conf/spark/standalone/hive-site.xml
@@ -205,6 +205,21 @@
 </property>
 
 <property>
+  <name>hive.prewarm.enabled</name>
+  <value>true</value>
+</property>
+
+<property>
+  <name>hive.prewarm.numcontainers</name>
+  <value>1</value>
+</property>
+
+<property>
+  <name>hive.prewarm.spark.timeout</name>
+  <value>30s</value>
+</property>
+
+<property>
   <name>spark.serializer</name>
   <value>org.apache.spark.serializer.KryoSerializer</value>
 </property>

http://git-wip-us.apache.org/repos/asf/hive/blob/a2c4aaed/data/conf/spark/yarn-client/hive-site.xml
----------------------------------------------------------------------
diff --git a/data/conf/spark/yarn-client/hive-site.xml b/data/conf/spark/yarn-client/hive-site.xml
index 9cda40d..bebca88 100644
--- a/data/conf/spark/yarn-client/hive-site.xml
+++ b/data/conf/spark/yarn-client/hive-site.xml
@@ -260,6 +260,16 @@
 </property>
 
 <property>
+  <name>hive.prewarm.enabled</name>
+  <value>true</value>
+</property>
+
+<property>
+  <name>hive.prewarm.spark.timeout</name>
+  <value>30s</value>
+</property>
+
+<property>
   <name>spark.testing</name>
   <value>true</value>
 </property>

http://git-wip-us.apache.org/repos/asf/hive/blob/a2c4aaed/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestUtil.java
----------------------------------------------------------------------
diff --git a/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestUtil.java b/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestUtil.java
index f6f35d3..b367732 100644
--- a/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestUtil.java
+++ b/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestUtil.java
@@ -1116,7 +1116,7 @@ public class QTestUtil {
     HiveConf.setVar(conf, HiveConf.ConfVars.HIVE_AUTHENTICATOR_MANAGER,
     "org.apache.hadoop.hive.ql.security.DummyAuthenticator");
     Utilities.clearWorkMap(conf);
-    CliSessionState ss = createSessionState();
+    CliSessionState ss = new CliSessionState(conf);
     assert ss != null;
     ss.in = System.in;
 
@@ -1176,33 +1176,6 @@ public class QTestUtil {
     return outf.getAbsolutePath();
   }
 
-  private CliSessionState createSessionState() {
-   return new CliSessionState(conf) {
-      @Override
-      public void setSparkSession(SparkSession sparkSession) {
-        super.setSparkSession(sparkSession);
-        if (sparkSession != null) {
-          try {
-            // Wait a little for cluster to init, at most 4 minutes
-            long endTime = System.currentTimeMillis() + 240000;
-            int expectedCores = conf.getInt("spark.executor.instances", 1) * 2;
-            while (sparkSession.getMemoryAndCores().getSecond() < expectedCores) {
-              if (System.currentTimeMillis() >= endTime) {
-                String msg = "Timed out waiting for Spark cluster to init";
-                throw new IllegalStateException(msg);
-              }
-              Thread.sleep(100);
-            }
-          } catch (Exception e) {
-            String msg = "Error trying to obtain executor info: " + e;
-            LOG.error(msg, e);
-            throw new IllegalStateException(msg, e);
-          }
-        }
-      }
-    };
-  }
-
   private CliSessionState startSessionState(boolean canReuseSession)
       throws IOException {
 
@@ -1211,7 +1184,7 @@ public class QTestUtil {
 
     String execEngine = conf.get("hive.execution.engine");
     conf.set("hive.execution.engine", "mr");
-    CliSessionState ss = createSessionState();
+    CliSessionState ss = new CliSessionState(conf);
     assert ss != null;
     ss.in = System.in;
     ss.out = System.out;

http://git-wip-us.apache.org/repos/asf/hive/blob/a2c4aaed/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/RemoteHiveSparkClient.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/RemoteHiveSparkClient.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/RemoteHiveSparkClient.java
index 78d5ff2..dbcf26a 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/RemoteHiveSparkClient.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/RemoteHiveSparkClient.java
@@ -76,7 +76,6 @@ public class RemoteHiveSparkClient implements HiveSparkClient {
   private static final String MR_JAR_PROPERTY = "tmpjars";
   private static final String MR_CREDENTIALS_LOCATION_PROPERTY = "mapreduce.job.credentials.binary";
   private static final transient Logger LOG = LoggerFactory.getLogger(RemoteHiveSparkClient.class);
-  private static final long MAX_PREWARM_TIME = 5000; // 5s
   private static final transient Splitter CSV_SPLITTER = Splitter.on(",").omitEmptyStrings();
 
   private transient Map<String, String> conf;
@@ -102,7 +101,8 @@ public class RemoteHiveSparkClient implements HiveSparkClient {
     remoteClient = SparkClientFactory.createClient(conf, hiveConf);
 
     if (HiveConf.getBoolVar(hiveConf, ConfVars.HIVE_PREWARM_ENABLED) &&
-        SparkClientUtilities.isYarnMaster(hiveConf.get("spark.master"))) {
+            (SparkClientUtilities.isYarnMaster(hiveConf.get("spark.master")) ||
+             SparkClientUtilities.isLocalMaster(hiveConf.get("spark.master")))) {
       int minExecutors = getExecutorsToWarm();
       if (minExecutors <= 0) {
         return;
@@ -110,12 +110,14 @@ public class RemoteHiveSparkClient implements HiveSparkClient {
 
       LOG.info("Prewarm Spark executors. The minimum number of executors to warm is " + minExecutors);
 
-      // Spend at most MAX_PREWARM_TIME to wait for executors to come up.
+      // Spend at most HIVE_PREWARM_SPARK_TIMEOUT to wait for executors to come up.
       int curExecutors = 0;
+      long maxPrewarmTime = HiveConf.getTimeVar(hiveConf, ConfVars.HIVE_PREWARM_SPARK_TIMEOUT,
+          TimeUnit.MILLISECONDS);
       long ts = System.currentTimeMillis();
       do {
         try {
-          curExecutors = getExecutorCount(MAX_PREWARM_TIME, TimeUnit.MILLISECONDS);
+          curExecutors = getExecutorCount(maxPrewarmTime, TimeUnit.MILLISECONDS);
         } catch (TimeoutException e) {
           // let's don't fail on future timeout since we have a timeout for pre-warm
           LOG.warn("Timed out getting executor count.", e);
@@ -125,9 +127,9 @@ public class RemoteHiveSparkClient implements HiveSparkClient {
           return;
         }
         Thread.sleep(500); // sleep half a second
-      } while (System.currentTimeMillis() - ts < MAX_PREWARM_TIME);
+      } while (System.currentTimeMillis() - ts < maxPrewarmTime);
 
-      LOG.info("Timeout (" + MAX_PREWARM_TIME / 1000 + "s) occurred while prewarming executors.
" +
+      LOG.info("Timeout (" + maxPrewarmTime / 1000 + "s) occurred while prewarming executors.
" +
           "The current number of executors is " + curExecutors);
     }
   }


Mime
View raw message