svn commit: r1696487 - in /pig/trunk: ./ src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/ src/org/apache/pig/backend/hadoop/executionengine/tez/util/ src/org/apache/pig/impl/util/

[ro...@apache.org]

Author: rohini
Date: Tue Aug 18 19:49:11 2015
New Revision: 1696487

URL: http://svn.apache.org/r1696487
Log:
PIG-4654: Reduce tez memory.reserve-fraction and clear spillables for better memory utilization (rohini)

Modified:
    pig/trunk/CHANGES.txt
    pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java
    pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java
    pig/trunk/src/org/apache/pig/impl/util/SpillableMemoryManager.java

Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1696487&r1=1696486&r2=1696487&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Tue Aug 18 19:49:11 2015
@@ -38,6 +38,8 @@ PIG-4639: Add better parser for Apache H
 
 BUG FIXES
 
+PIG-4654: Reduce tez memory.reserve-fraction and clear spillables for better memory utilization (rohini)
+
 PIG-4628: Pig 0.14 job with order by fails in mapreduce mode with Oozie (knoguchi)
 
 PIG-4651: Optimize NullablePartitionWritable serialization for skewed join (rohini)

Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java?rev=1696487&r1=1696486&r2=1696487&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java Tue Aug 18 19:49:11 2015
@@ -193,6 +193,8 @@ public class PigProcessor extends Abstra
                 leaf = leaves.get(0);
             }
 
+            LOG.info("Aliases being processed per job phase (AliasName[line,offset]): " + conf.get("pig.alias.location"));
+
             runPipeline(leaf);
 
             // Calling EvalFunc.finish()

Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java?rev=1696487&r1=1696486&r2=1696487&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/util/MRToTezHelper.java Tue Aug 18 19:49:11 2015
@@ -180,6 +180,9 @@ public class MRToTezHelper {
                     tezConf.get(MRJobConfig.JOB_CANCEL_DELEGATION_TOKEN));
         }
 
+        // Hardcoding at AM level instead of setting per vertex till TEZ-2710 is available
+        dagAMConf.setIfUnset(TezConfiguration.TEZ_TASK_SCALE_MEMORY_RESERVE_FRACTION, "0.5");
+
         removeUnwantedMRSettings(dagAMConf);
 
         return dagAMConf;

Modified: pig/trunk/src/org/apache/pig/impl/util/SpillableMemoryManager.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/impl/util/SpillableMemoryManager.java?rev=1696487&r1=1696486&r2=1696487&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/impl/util/SpillableMemoryManager.java (original)
+++ pig/trunk/src/org/apache/pig/impl/util/SpillableMemoryManager.java Tue Aug 18 19:49:11 2015
@@ -36,6 +36,8 @@ import javax.management.openmbean.Compos
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.pig.JVMReuseManager;
+import org.apache.pig.StaticDataCleanup;
 
 /**
  * This class Tracks the tenured pool and a list of Spillable objects. When memory gets low, this
@@ -48,7 +50,7 @@ import org.apache.commons.logging.LogFac
  */
 public class SpillableMemoryManager implements NotificationListener {
 
-    private final Log log = LogFactory.getLog(getClass());
+    private static final Log log = LogFactory.getLog(SpillableMemoryManager.class);
 
     private LinkedList<WeakReference<Spillable>> spillables = new LinkedList<WeakReference<Spillable>>();
     // References to spillables with size
@@ -89,7 +91,17 @@ public class SpillableMemoryManager impl
 
     private volatile boolean blockRegisterOnSpill = false;
 
-    private static volatile SpillableMemoryManager manager = new SpillableMemoryManager();
+    private static final SpillableMemoryManager manager = new SpillableMemoryManager();
+
+    static {
+        JVMReuseManager.getInstance().registerForStaticDataCleanup(SpillableMemoryManager.class);
+    }
+
+    @StaticDataCleanup
+    public static void cleanupStaticData() {
+        manager.spillables.clear();
+        manager.accumulatedFreeSize = 0L;
+    }
 
     private SpillableMemoryManager() {
         ((NotificationEmitter)ManagementFactory.getMemoryMXBean()).addNotificationListener(this, null, null);