aboutsummaryrefslogtreecommitdiff
path: root/python/pyspark/serializers.py
diff options
context:
space:
mode:
authorDavies Liu <davies.liu@gmail.com>2014-07-24 22:53:47 -0700
committerMatei Zaharia <matei@databricks.com>2014-07-24 22:53:47 -0700
commit14174abd421318e71c16edd24224fd5094bdfed4 (patch)
tree0af1cd49e09c80d8b7dd62d9de90ff03842db382 /python/pyspark/serializers.py
parenteff9714e1c88e39e28317358ca9ec87677f121dc (diff)
downloadspark-14174abd421318e71c16edd24224fd5094bdfed4.tar.gz
spark-14174abd421318e71c16edd24224fd5094bdfed4.tar.bz2
spark-14174abd421318e71c16edd24224fd5094bdfed4.zip
[SPARK-2538] [PySpark] Hash based disk spilling aggregation
During aggregation in Python worker, if the memory usage is above spark.executor.memory, it will do disk spilling aggregation. It will split the aggregation into multiple stage, in each stage, it will partition the aggregated data by hash and dump them into disks. After all the data are aggregated, it will merge all the stages together (partition by partition). Author: Davies Liu <davies.liu@gmail.com> Closes #1460 from davies/spill and squashes the following commits: cad91bf [Davies Liu] call gc.collect() after data.clear() to release memory as much as possible. 37d71f7 [Davies Liu] balance the partitions 902f036 [Davies Liu] add shuffle.py into run-tests dcf03a9 [Davies Liu] fix memory_info() of psutil 67e6eba [Davies Liu] comment for MAX_TOTAL_PARTITIONS f6bd5d6 [Davies Liu] rollback next_limit() again, the performance difference is huge: e74b785 [Davies Liu] fix code style and change next_limit to memory_limit 400be01 [Davies Liu] address all the comments 6178844 [Davies Liu] refactor and improve docs fdd0a49 [Davies Liu] add long doc string for ExternalMerger 1a97ce4 [Davies Liu] limit used memory and size of objects in partitionBy() e6cc7f9 [Davies Liu] Merge branch 'master' into spill 3652583 [Davies Liu] address comments e78a0a0 [Davies Liu] fix style 24cec6a [Davies Liu] get local directory by SPARK_LOCAL_DIR 57ee7ef [Davies Liu] update docs 286aaff [Davies Liu] let spilled aggregation in Python configurable e9a40f6 [Davies Liu] recursive merger 6edbd1f [Davies Liu] Hash based disk spilling aggregation
Diffstat (limited to 'python/pyspark/serializers.py')
-rw-r--r--python/pyspark/serializers.py29
1 files changed, 28 insertions, 1 deletions
diff --git a/python/pyspark/serializers.py b/python/pyspark/serializers.py
index 9be78b39fb..03b31ae962 100644
--- a/python/pyspark/serializers.py
+++ b/python/pyspark/serializers.py
@@ -193,7 +193,7 @@ class BatchedSerializer(Serializer):
return chain.from_iterable(self._load_stream_without_unbatching(stream))
def _load_stream_without_unbatching(self, stream):
- return self.serializer.load_stream(stream)
+ return self.serializer.load_stream(stream)
def __eq__(self, other):
return (isinstance(other, BatchedSerializer) and
@@ -302,6 +302,33 @@ class MarshalSerializer(FramedSerializer):
loads = marshal.loads
+class AutoSerializer(FramedSerializer):
+ """
+ Choose marshal or cPickle as serialization protocol autumatically
+ """
+ def __init__(self):
+ FramedSerializer.__init__(self)
+ self._type = None
+
+ def dumps(self, obj):
+ if self._type is not None:
+ return 'P' + cPickle.dumps(obj, -1)
+ try:
+ return 'M' + marshal.dumps(obj)
+ except Exception:
+ self._type = 'P'
+ return 'P' + cPickle.dumps(obj, -1)
+
+ def loads(self, obj):
+ _type = obj[0]
+ if _type == 'M':
+ return marshal.loads(obj[1:])
+ elif _type == 'P':
+ return cPickle.loads(obj[1:])
+ else:
+ raise ValueError("invalid sevialization type: %s" % _type)
+
+
class UTF8Deserializer(Serializer):
"""
Deserializes streams written by String.getBytes.