svn commit: r1651265 - in /hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec: Utilities.java spark/SparkTask.java

[xu...@apache.org]

Author: xuefu
Date: Tue Jan 13 02:16:01 2015
New Revision: 1651265

URL: http://svn.apache.org/r1651265
Log:
HIVE-9135: Cache Map and Reduce works in RSC [Spark Branch] (Jimmy via Xuefu)

Modified:
    hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
    hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkTask.java

Modified: hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
URL: http://svn.apache.org/viewvc/hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java?rev=1651265&r1=1651264&r2=1651265&view=diff
==============================================================================
--- hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java (original)
+++ hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java Tue Jan 13 02:16:01 2015
@@ -31,7 +31,6 @@ import java.io.ByteArrayOutputStream;
 import java.io.DataInput;
 import java.io.EOFException;
 import java.io.File;
-import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
@@ -236,8 +235,13 @@ public final class Utilities {
     // prevent instantiation
   }
 
-  private static Map<Path, BaseWork> gWorkMap = Collections
-      .synchronizedMap(new HashMap<Path, BaseWork>());
+  private static ThreadLocal<Map<Path, BaseWork>> gWorkMap =
+      new ThreadLocal<Map<Path, BaseWork>>() {
+    protected Map<Path, BaseWork> initialValue() {
+      return new HashMap<Path, BaseWork>();
+    }
+  };
+
   private static final String CLASS_NAME = Utilities.class.getName();
   private static final Log LOG = LogFactory.getLog(CLASS_NAME);
 
@@ -344,7 +348,7 @@ public final class Utilities {
    */
   public static void setBaseWork(Configuration conf, String name, BaseWork work) {
     Path path = getPlanPath(conf, name);
-    gWorkMap.put(path, work);
+    gWorkMap.get().put(path, work);
   }
 
   /**
@@ -356,15 +360,14 @@ public final class Utilities {
    * @throws RuntimeException if the configuration files are not proper or if plan can not be loaded
    */
   private static BaseWork getBaseWork(Configuration conf, String name) {
-    BaseWork gWork = null;
     Path path = null;
     InputStream in = null;
     try {
       path = getPlanPath(conf, name);
       LOG.info("PLAN PATH = " + path);
       assert path != null;
-      if (!gWorkMap.containsKey(path)
-        || HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_EXECUTION_ENGINE).equals("spark")) {
+      BaseWork gWork = gWorkMap.get().get(path);
+      if (gWork == null) {
         Path localPath;
         if (conf.getBoolean("mapreduce.task.uberized", false) && name.equals(REDUCE_PLAN_NAME)) {
           localPath = new Path(name);
@@ -414,10 +417,9 @@ public final class Utilities {
         } else if (name.contains(MERGE_PLAN_NAME)) {
           gWork = deserializePlan(in, MapWork.class, conf);
         }
-        gWorkMap.put(path, gWork);
-      } else {
+        gWorkMap.get().put(path, gWork);
+      } else if (LOG.isDebugEnabled()) {
         LOG.debug("Found plan in cache for name: " + name);
-        gWork = gWorkMap.get(path);
       }
       return gWork;
     } catch (FileNotFoundException fnf) {
@@ -709,7 +711,7 @@ public final class Utilities {
       }
 
       // Cache the plan in this process
-      gWorkMap.put(planPath, w);
+      gWorkMap.get().put(planPath, w);
       return planPath;
     } catch (Exception e) {
       String msg = "Error caching " + name + ": " + e;
@@ -3629,15 +3631,15 @@ public final class Utilities {
     Path mapPath = getPlanPath(conf, MAP_PLAN_NAME);
     Path reducePath = getPlanPath(conf, REDUCE_PLAN_NAME);
     if (mapPath != null) {
-      gWorkMap.remove(mapPath);
+      gWorkMap.get().remove(mapPath);
     }
     if (reducePath != null) {
-      gWorkMap.remove(reducePath);
+      gWorkMap.get().remove(reducePath);
     }
   }
 
   public static void clearWorkMap() {
-    gWorkMap.clear();
+    gWorkMap.get().clear();
   }
 
   /**

Modified: hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkTask.java
URL: http://svn.apache.org/viewvc/hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkTask.java?rev=1651265&r1=1651264&r2=1651265&view=diff
==============================================================================
--- hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkTask.java (original)
+++ hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkTask.java Tue Jan 13 02:16:01 2015
@@ -124,6 +124,7 @@ public class SparkTask extends Task<Spar
       LOG.error(msg, e);
       rc = 1;
     } finally {
+      Utilities.clearWork(conf);
       if (sparkSession != null && sparkSessionManager != null) {
         rc = close(rc);
         try {