fix code style and change next_limit to memory_limit

Author: davies

python/pyspark/rdd.py

@@ -1229,7 +1229,7 @@ def partitionBy(self, numPartitions, partitionFunc=portable_hash):
         # Instead, we'll form the hash buckets in Python,
         # transferring O(numPartitions) objects to Java.
         # Each object is a (splitNumber, [objects]) pair.
-        # In order to void too huge objects, the objects are
+        # In order to avoid too huge objects, the objects are
         # grouped into chunks.
         outputSerializer = self.ctx._unbatched_serializer
 
@@ -1316,11 +1316,11 @@ def combineByKey(self, createCombiner, mergeValue, mergeCombiners,
         spill = (self.ctx._conf.get("spark.shuffle.spill", 'True').lower()
                  == 'true')
         memory = (_parse_memory(self.ctx._conf.get(
-                    "spark.python.worker.memory","512m")
+                    "spark.python.worker.memory", "512m")
         agg = Aggregator(createCombiner, mergeValue, mergeCombiners)
 
         def combineLocally(iterator):
-            merger = ExternalMerger(agg, memory, serializer) \
+            merger = ExternalMerger(agg, memory * 0.9, serializer) \
                          if spill else InMemoryMerger(agg)
             merger.mergeValues(iterator)
             return merger.iteritems()

python/pyspark/shuffle.py

@@ -216,13 +216,14 @@ def _get_spill_dir(self, n):
         """ Choose one directory for spill by number n """
         return os.path.join(self.localdirs[n % len(self.localdirs)], str(n))
 
-    def next_limit(self):
-        """
-        Return the next memory limit. If the memory is not released
-        after spilling, it will dump the data only when the used memory
-        starts to increase.
-        """
-        return max(self.memory_limit, get_used_memory() * 1.05)
+    def _next_limit(self):
+        #"""
+        #Return the next memory limit. If the memory is not released
+        #after spilling, it will dump the data only when the used memory
+        #starts to increase.
+        #"""
+        #return max(self.memory_limit, get_used_memory() * 1.05)
+        return self.memory_limit
 
     def mergeValues(self, iterator):
         """ Combine the items by creator and combiner """
@@ -237,7 +238,7 @@ def mergeValues(self, iterator):
             c += 1
             if c % batch == 0 and get_used_memory() > self.memory_limit:
                 self._first_spill()
-                self._partitioned_mergeValues(iterator, self.next_limit())
+                self._partitioned_mergeValues(iterator, self._next_limit())
                 break
 
     def _partition(self, key):
@@ -259,7 +260,7 @@ def _partitioned_mergeValues(self, iterator, limit=0):
             c += 1
             if c % batch == 0 and get_used_memory() > limit:
                 self._spill()
-                limit = self.next_limit()
+                limit = self._next_limit()
 
     def mergeCombiners(self, iterator, check=True):
         """ Merge (K,V) pair by mergeCombiner """
@@ -275,7 +276,7 @@ def mergeCombiners(self, iterator, check=True):
             c += 1
             if c % batch == 0 and get_used_memory() > self.memory_limit:
                 self._first_spill()
-                self._partitioned_mergeCombiners(iterator, self.next_limit())
+                self._partitioned_mergeCombiners(iterator, self._next_limit())
                 break
 
     def _partitioned_mergeCombiners(self, iterator, limit=0):
@@ -291,7 +292,7 @@ def _partitioned_mergeCombiners(self, iterator, limit=0):
             c += 1
             if c % self.batch == 0 and get_used_memory() > limit:
                 self._spill()
-                limit = self.next_limit()
+                limit = self._next_limit()
 
     def _first_spill(self):
         """
@@ -349,7 +350,7 @@ def _external_items(self):
         assert not self.data
         if any(self.pdata):
             self._spill()
-        hard_limit = self.next_limit()
+        hard_limit = self._next_limit()
 
         try:
             for i in range(self.partitions):
@@ -406,7 +407,7 @@ def _recursive_merged_items(self, start):
             m = ExternalMerger(self.agg, self.memory_limit, self.serializer,
                     subdirs, self.scale * self.partitions)
             m.pdata = [{} for _ in range(self.partitions)]
-            limit = self.next_limit()
+            limit = self._next_limit()
 
             for j in range(self.spills):
                 path = self._get_spill_dir(j)
@@ -416,7 +417,7 @@ def _recursive_merged_items(self, start):
 
                 if get_used_memory() > limit:
                     m._spill()
-                    limit = self.next_limit()
+                    limit = self._next_limit()
 
             for v in m._external_items():
                 yield v