From 33f34614b0620835b55924c3bc486bc7add6e906 Mon Sep 17 00:00:00 2001 From: Eric Liang Date: Fri, 9 Sep 2016 09:52:49 -0700 Subject: [PATCH 1/5] Fri Sep 9 09:52:49 PDT 2016 --- python/pyspark/broadcast.py | 8 +++++++- python/pyspark/cloudpickle.py | 9 +++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/python/pyspark/broadcast.py b/python/pyspark/broadcast.py index a0b819220e6d3..8dd2e1fde71bf 100644 --- a/python/pyspark/broadcast.py +++ b/python/pyspark/broadcast.py @@ -75,7 +75,13 @@ def __init__(self, sc=None, value=None, pickle_registry=None, path=None): self._path = path def dump(self, value, f): - pickle.dump(value, f, 2) + try: + pickle.dump(value, f, 2) + except pickle.PickleError: + raise + except Exception as e: + msg = "Could not serialize broadcast: " + e.__class__.__name__ + ": " + e.message + raise pickle.PicklingError, pickle.PicklingError(msg), sys.exc_info()[2] f.close() return f.name diff --git a/python/pyspark/cloudpickle.py b/python/pyspark/cloudpickle.py index 822ae46e45111..a254bacecd43c 100644 --- a/python/pyspark/cloudpickle.py +++ b/python/pyspark/cloudpickle.py @@ -109,6 +109,15 @@ def dump(self, obj): if 'recursion' in e.args[0]: msg = """Could not pickle object as excessively deep recursion required.""" raise pickle.PicklingError(msg) + except pickle.PickleError: + raise + except Exception as e: + if "'i' format requires" in e.message: + msg = "Object too large to serialize: " + e.message + else: + msg = "Could not serialize object: " + e.__class__.__name__ + ": " + e.message + raise pickle.PicklingError, pickle.PicklingError(msg), sys.exc_info()[2] + def save_memoryview(self, obj): """Fallback to save_string""" From df0ecc4284b6b1faeb7816fd35347d2a101f933d Mon Sep 17 00:00:00 2001 From: Eric Liang Date: Fri, 9 Sep 2016 12:29:15 -0700 Subject: [PATCH 2/5] Fri Sep 9 12:29:15 PDT 2016 --- python/pyspark/broadcast.py | 2 +- python/pyspark/cloudpickle.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/python/pyspark/broadcast.py b/python/pyspark/broadcast.py index 8dd2e1fde71bf..ced0cb0e1b4e7 100644 --- a/python/pyspark/broadcast.py +++ b/python/pyspark/broadcast.py @@ -81,7 +81,7 @@ def dump(self, value, f): raise except Exception as e: msg = "Could not serialize broadcast: " + e.__class__.__name__ + ": " + e.message - raise pickle.PicklingError, pickle.PicklingError(msg), sys.exc_info()[2] + raise pickle.PicklingError, pickle.PicklingError(msg), sys.exc_info()[2] # noqa f.close() return f.name diff --git a/python/pyspark/cloudpickle.py b/python/pyspark/cloudpickle.py index a254bacecd43c..2c66f7cdc80b2 100644 --- a/python/pyspark/cloudpickle.py +++ b/python/pyspark/cloudpickle.py @@ -116,7 +116,7 @@ def dump(self, obj): msg = "Object too large to serialize: " + e.message else: msg = "Could not serialize object: " + e.__class__.__name__ + ": " + e.message - raise pickle.PicklingError, pickle.PicklingError(msg), sys.exc_info()[2] + raise pickle.PicklingError, pickle.PicklingError(msg), sys.exc_info()[2] # noqa def save_memoryview(self, obj): From fdaed3fdfa5d47a3cba03504a82ce94b26b7eae6 Mon Sep 17 00:00:00 2001 From: Eric Liang Date: Fri, 9 Sep 2016 18:02:47 -0700 Subject: [PATCH 3/5] drop stack --- python/pyspark/broadcast.py | 2 +- python/pyspark/cloudpickle.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/python/pyspark/broadcast.py b/python/pyspark/broadcast.py index ced0cb0e1b4e7..973c790299dd9 100644 --- a/python/pyspark/broadcast.py +++ b/python/pyspark/broadcast.py @@ -81,7 +81,7 @@ def dump(self, value, f): raise except Exception as e: msg = "Could not serialize broadcast: " + e.__class__.__name__ + ": " + e.message - raise pickle.PicklingError, pickle.PicklingError(msg), sys.exc_info()[2] # noqa + raise pickle.PicklingError(msg) f.close() return f.name diff --git a/python/pyspark/cloudpickle.py b/python/pyspark/cloudpickle.py index 2c66f7cdc80b2..dddd5d3e3df1a 100644 --- a/python/pyspark/cloudpickle.py +++ b/python/pyspark/cloudpickle.py @@ -116,7 +116,7 @@ def dump(self, obj): msg = "Object too large to serialize: " + e.message else: msg = "Could not serialize object: " + e.__class__.__name__ + ": " + e.message - raise pickle.PicklingError, pickle.PicklingError(msg), sys.exc_info()[2] # noqa + raise pickle.PicklingError(msg) def save_memoryview(self, obj): From 3d2b445dafd5e92ef1e9fb69e4d58f12549c1dfc Mon Sep 17 00:00:00 2001 From: Eric Liang Date: Tue, 13 Sep 2016 15:32:05 -0700 Subject: [PATCH 4/5] Tue Sep 13 15:32:05 PDT 2016 --- python/pyspark/broadcast.py | 2 ++ python/pyspark/cloudpickle.py | 1 + 2 files changed, 3 insertions(+) diff --git a/python/pyspark/broadcast.py b/python/pyspark/broadcast.py index 973c790299dd9..d3e775b7376fc 100644 --- a/python/pyspark/broadcast.py +++ b/python/pyspark/broadcast.py @@ -18,6 +18,7 @@ import os import sys import gc +from cloudpickle import print_exec from tempfile import NamedTemporaryFile if sys.version < '3': @@ -81,6 +82,7 @@ def dump(self, value, f): raise except Exception as e: msg = "Could not serialize broadcast: " + e.__class__.__name__ + ": " + e.message + print_exec(sys.stderr) raise pickle.PicklingError(msg) f.close() return f.name diff --git a/python/pyspark/cloudpickle.py b/python/pyspark/cloudpickle.py index dddd5d3e3df1a..da2b2f3757967 100644 --- a/python/pyspark/cloudpickle.py +++ b/python/pyspark/cloudpickle.py @@ -116,6 +116,7 @@ def dump(self, obj): msg = "Object too large to serialize: " + e.message else: msg = "Could not serialize object: " + e.__class__.__name__ + ": " + e.message + print_exec(sys.stderr) raise pickle.PicklingError(msg) From 913f2f4dcb26ab63a869f393f62c2ef357ba958e Mon Sep 17 00:00:00 2001 From: Eric Liang Date: Wed, 14 Sep 2016 11:37:14 -0700 Subject: [PATCH 5/5] Wed Sep 14 11:37:14 PDT 2016 --- python/pyspark/broadcast.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/python/pyspark/broadcast.py b/python/pyspark/broadcast.py index d3e775b7376fc..74dee1420754a 100644 --- a/python/pyspark/broadcast.py +++ b/python/pyspark/broadcast.py @@ -18,9 +18,10 @@ import os import sys import gc -from cloudpickle import print_exec from tempfile import NamedTemporaryFile +from pyspark.cloudpickle import print_exec + if sys.version < '3': import cPickle as pickle else: