hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chengxi...@apache.org
Subject hive git commit: HIVE-10550: Dynamic RDD caching optimization for HoS [Spark Branch] (Chengxiang reviewed by Xuefu)
Date Thu, 28 May 2015 05:26:23 GMT
Repository: hive
Updated Branches:
  refs/heads/spark a318d5b00 -> 461008464


HIVE-10550: Dynamic RDD caching optimization for HoS [Spark Branch] (Chengxiang reviewed by
Xuefu)


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

Branch: refs/heads/spark
Commit: 461008464053f75592d737cdc130b7f0c8e4e4c3
Parents: a318d5b
Author: chengxiang <chengxiang@apache.com>
Authored: Thu May 28 13:31:26 2015 +0800
Committer: chengxiang <chengxiang@apache.com>
Committed: Thu May 28 13:31:26 2015 +0800

----------------------------------------------------------------------
 .../hadoop/hive/ql/exec/spark/CacheTran.java    | 54 ++++++++++++++++++++
 .../hadoop/hive/ql/exec/spark/MapTran.java      | 17 +++---
 .../hadoop/hive/ql/exec/spark/ReduceTran.java   | 17 +++---
 .../hadoop/hive/ql/exec/spark/SparkPlan.java    |  3 ++
 .../hive/ql/exec/spark/SparkPlanGenerator.java  | 25 +++++++--
 .../hive/ql/exec/spark/SparkUtilities.java      | 36 +++++++++++++
 6 files changed, 134 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/46100846/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/CacheTran.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/CacheTran.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/CacheTran.java
new file mode 100644
index 0000000..5ec27ec
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/CacheTran.java
@@ -0,0 +1,54 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.exec.spark;
+
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.storage.StorageLevel;
+
+public abstract class CacheTran<KI extends WritableComparable, VI, KO extends WritableComparable,
VO>
+  implements SparkTran<KI, VI, KO, VO> {
+  // whether to cache current RDD.
+  private boolean caching = false;
+  private JavaPairRDD<KO, VO> cachedRDD;
+
+  protected CacheTran(boolean cache) {
+    this.caching = cache;
+  }
+
+  @Override
+  public JavaPairRDD<KO, VO> transform(
+    JavaPairRDD<KI, VI> input) {
+    if (caching) {
+      if (cachedRDD == null) {
+        cachedRDD = doTransform(input);
+        cachedRDD.persist(StorageLevel.MEMORY_AND_DISK());
+      }
+      return cachedRDD;
+    } else {
+      return doTransform(input);
+    }
+  }
+
+  public Boolean isCacheEnable() {
+    return caching;
+  }
+
+  protected abstract JavaPairRDD<KO, VO> doTransform(JavaPairRDD<KI, VI> input);
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/46100846/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/MapTran.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/MapTran.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/MapTran.java
index 2170243..2a18991 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/MapTran.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/MapTran.java
@@ -22,12 +22,20 @@ import org.apache.hadoop.hive.ql.io.HiveKey;
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.spark.api.java.JavaPairRDD;
 
-public class MapTran implements SparkTran<BytesWritable, BytesWritable, HiveKey, BytesWritable>
{
+public class MapTran extends CacheTran<BytesWritable, BytesWritable, HiveKey, BytesWritable>
{
   private HiveMapFunction mapFunc;
   private String name = "MapTran";
 
+  public MapTran() {
+    this(false);
+  }
+
+  public MapTran(boolean cache) {
+    super(cache);
+  }
+
   @Override
-  public JavaPairRDD<HiveKey, BytesWritable> transform(
+  public JavaPairRDD<HiveKey, BytesWritable> doTransform(
       JavaPairRDD<BytesWritable, BytesWritable> input) {
     return input.mapPartitionsToPair(mapFunc);
   }
@@ -42,11 +50,6 @@ public class MapTran implements SparkTran<BytesWritable, BytesWritable,
HiveKey,
   }
 
   @Override
-  public Boolean isCacheEnable() {
-    return null;
-  }
-
-  @Override
   public void setName(String name) {
     this.name = name;
   }

http://git-wip-us.apache.org/repos/asf/hive/blob/46100846/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/ReduceTran.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/ReduceTran.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/ReduceTran.java
index e60dfac..3d56876 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/ReduceTran.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/ReduceTran.java
@@ -22,12 +22,20 @@ import org.apache.hadoop.hive.ql.io.HiveKey;
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.spark.api.java.JavaPairRDD;
 
-public class ReduceTran implements SparkTran<HiveKey, Iterable<BytesWritable>, HiveKey,
BytesWritable> {
+public class ReduceTran extends CacheTran<HiveKey, Iterable<BytesWritable>, HiveKey,
BytesWritable> {
   private HiveReduceFunction reduceFunc;
   private String name = "Reduce";
 
+  public ReduceTran() {
+    this(false);
+  }
+
+  public ReduceTran(boolean caching) {
+    super(caching);
+  }
+
   @Override
-  public JavaPairRDD<HiveKey, BytesWritable> transform(
+  public JavaPairRDD<HiveKey, BytesWritable> doTransform(
       JavaPairRDD<HiveKey, Iterable<BytesWritable>> input) {
     return input.mapPartitionsToPair(reduceFunc);
   }
@@ -42,11 +50,6 @@ public class ReduceTran implements SparkTran<HiveKey, Iterable<BytesWritable>,
H
   }
 
   @Override
-  public Boolean isCacheEnable() {
-    return null;
-  }
-
-  @Override
   public void setName(String name) {
     this.name = name;
   }

http://git-wip-us.apache.org/repos/asf/hive/blob/46100846/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkPlan.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkPlan.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkPlan.java
index ee5c78a..762f734 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkPlan.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkPlan.java
@@ -88,6 +88,9 @@ public class SparkPlan {
     }
 
     perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.SPARK_BUILD_RDD_GRAPH);
+    if (LOG.isDebugEnabled()) {
+      LOG.info("print generated spark rdd graph:\n" + SparkUtilities.rddGraphToString(finalRDD));
+    }
     return finalRDD;
   }
 

http://git-wip-us.apache.org/repos/asf/hive/blob/46100846/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkPlanGenerator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkPlanGenerator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkPlanGenerator.java
index 3f240f5..3a1eff8 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkPlanGenerator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkPlanGenerator.java
@@ -102,7 +102,7 @@ public class SparkPlanGenerator {
     try {
       for (BaseWork work : sparkWork.getAllWork()) {
         perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.SPARK_CREATE_TRAN + work.getName());
-        SparkTran tran = generate(work);
+        SparkTran tran = generate(work, sparkWork);
         SparkTran parentTran = generateParentTran(sparkPlan, sparkWork, work);
         sparkPlan.addTran(tran);
         sparkPlan.connect(parentTran, tran);
@@ -206,18 +206,19 @@ public class SparkPlanGenerator {
     return new ShuffleTran(sparkPlan, shuffler, edge.getNumPartitions(), toCache);
   }
 
-  private SparkTran generate(BaseWork work) throws Exception {
+  private SparkTran generate(BaseWork work, SparkWork sparkWork) throws Exception {
     initStatsPublisher(work);
     JobConf newJobConf = cloneJobConf(work);
     checkSpecs(work, newJobConf);
     byte[] confBytes = KryoSerializer.serializeJobConf(newJobConf);
+    boolean caching = isCachingWork(work, sparkWork);
     if (work instanceof MapWork) {
-      MapTran mapTran = new MapTran();
+      MapTran mapTran = new MapTran(caching);
       HiveMapFunction mapFunc = new HiveMapFunction(confBytes, sparkReporter);
       mapTran.setMapFunction(mapFunc);
       return mapTran;
     } else if (work instanceof ReduceWork) {
-      ReduceTran reduceTran = new ReduceTran();
+      ReduceTran reduceTran = new ReduceTran(caching);
       HiveReduceFunction reduceFunc = new HiveReduceFunction(confBytes, sparkReporter);
       reduceTran.setReduceFunction(reduceFunc);
       return reduceTran;
@@ -227,6 +228,22 @@ public class SparkPlanGenerator {
     }
   }
 
+  private boolean isCachingWork(BaseWork work, SparkWork sparkWork) {
+    boolean caching = true;
+    List<BaseWork> children = sparkWork.getChildren(work);
+    if (children.size() < 2) {
+      caching = false;
+    } else {
+      // do not cache this if its child RDD is intend to be cached.
+      for (BaseWork child : children) {
+        if (cloneToWork.containsKey(child)) {
+          caching = false;
+        }
+      }
+    }
+    return caching;
+  }
+
   private void checkSpecs(BaseWork work, JobConf jc) throws Exception {
     Set<Operator<?>> opList = work.getAllOperators();
     for (Operator<?> op : opList) {

http://git-wip-us.apache.org/repos/asf/hive/blob/46100846/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkUtilities.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkUtilities.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkUtilities.java
index e6c845c..ca0ffb6 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkUtilities.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkUtilities.java
@@ -21,6 +21,7 @@ import java.io.File;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.Collection;
 import java.util.UUID;
 
 import org.apache.commons.io.FilenameUtils;
@@ -34,7 +35,12 @@ import org.apache.hadoop.hive.ql.io.HiveKey;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
 import org.apache.hadoop.hive.ql.session.SessionState;
 import org.apache.hadoop.io.BytesWritable;
+import org.apache.spark.Dependency;
 import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.rdd.RDD;
+import org.apache.spark.rdd.UnionRDD;
+import scala.collection.JavaConversions;
 
 /**
  * Contains utilities methods used as part of Spark tasks.
@@ -122,4 +128,34 @@ public class SparkUtilities {
     SessionState.get().setSparkSession(sparkSession);
     return sparkSession;
   }
+
+
+  public static String rddGraphToString(JavaPairRDD rdd) {
+    StringBuilder sb = new StringBuilder();
+    rddToString(rdd.rdd(), sb, "");
+    return sb.toString();
+  }
+
+  private static void rddToString(RDD rdd, StringBuilder sb, String offset) {
+    sb.append(offset).append(rdd.getClass().getCanonicalName()).append("[").append(rdd.hashCode()).append("]");
+    if (rdd.getStorageLevel().useMemory()) {
+      sb.append("(cached)");
+    }
+    sb.append("\n");
+    Collection<Dependency> dependencies = JavaConversions.asJavaCollection(rdd.dependencies());
+    if (dependencies != null) {
+      offset += "\t";
+      for (Dependency dependency : dependencies) {
+        RDD parentRdd = dependency.rdd();
+        rddToString(parentRdd, sb, offset);
+      }
+    } else if (rdd instanceof UnionRDD) {
+      UnionRDD unionRDD = (UnionRDD) rdd;
+      offset += "\t";
+      Collection<RDD> parentRdds = JavaConversions.asJavaCollection(unionRDD.rdds());
+      for (RDD parentRdd : parentRdds) {
+        rddToString(parentRdd, sb, offset);
+      }
+    }
+  }
 }


Mime
View raw message