svn commit: r1771840 - in /pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark: SparkLauncher.java SparkUtil.java

[xu...@apache.org]

Author: xuefu
Date: Tue Nov 29 04:01:46 2016
New Revision: 1771840

URL: http://svn.apache.org/viewvc?rev=1771840&view=rev
Log:
PIG-5068: Set SPARK_REDUCERS by pig.properties not by system configuration (Liyun via Xuefu)

Modified:
    pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkLauncher.java
    pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkUtil.java

Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkLauncher.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkLauncher.java?rev=1771840&r1=1771839&r2=1771840&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkLauncher.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkLauncher.java Tue Nov 29 04:01:46 2016
@@ -650,5 +650,9 @@ public class SparkLauncher extends Launc
         SchemaTupleBackend.initialize(jobConf, pigContext);
         Utils.setDefaultTimeZone(jobConf);
         PigMapReduce.sJobConfInternal.set(jobConf);
+        String sparkReducers = pigContext.getProperties().getProperty("spark.reducers");
+        if (sparkReducers != null) {
+            SparkUtil.setSparkReducers(Integer.parseInt(sparkReducers));
+        }
     }
 }

Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkUtil.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkUtil.java?rev=1771840&r1=1771839&r2=1771840&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkUtil.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/SparkUtil.java Tue Nov 29 04:01:46 2016
@@ -25,7 +25,6 @@ import java.util.Random;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 
-import org.apache.spark.broadcast.Broadcast;
 import scala.Product2;
 import scala.Tuple2;
 import scala.collection.JavaConversions;
@@ -55,10 +54,12 @@ import org.apache.pig.impl.plan.PlanExce
 import org.apache.pig.impl.util.ObjectSerializer;
 import org.apache.spark.HashPartitioner;
 import org.apache.spark.Partitioner;
+import org.apache.spark.broadcast.Broadcast;
 import org.apache.spark.rdd.RDD;
 
 public class SparkUtil {
 
+    private static ThreadLocal<Integer> SPARK_REDUCERS = null;
     private static ConcurrentHashMap<String, Broadcast<List<Tuple>>> broadcastedVars = new ConcurrentHashMap() ;
 
     public static <T> ClassTag<T> getManifest(Class<T> clazz) {
@@ -130,10 +131,8 @@ public class SparkUtil {
 
     public static int getParallelism(List<RDD<Tuple>> predecessors,
             PhysicalOperator physicalOperator) {
-
-        String numReducers = System.getenv("SPARK_REDUCERS");
-        if (numReducers != null) {
-            return Integer.parseInt(numReducers);
+        if (SPARK_REDUCERS != null) {
+            return getSparkReducers();
         }
 
         int parallelism = physicalOperator.getRequestedParallelism();
@@ -178,4 +177,13 @@ public class SparkUtil {
     static public ConcurrentHashMap<String, Broadcast<List<Tuple>>> getBroadcastedVars() {
         return broadcastedVars;
     }
+
+
+    public static int getSparkReducers() {
+        return SPARK_REDUCERS.get();
+    }
+
+    public static void setSparkReducers(int sparkReducers) {
+        SPARK_REDUCERS.set(sparkReducers);
+    }
 }
\ No newline at end of file