Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[SPARK-10542] [PYSPARK] fix serialize namedtuple #8707

Closed
wants to merge 4 commits into from
Closed

[SPARK-10542] [PYSPARK] fix serialize namedtuple #8707

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d7ef6ce
fix serialize namedtuple
Sep 10, 2015
1d766aa
use special case for namedtuple
Sep 10, 2015
a2f9f36
fix test in python 3.4
Sep 10, 2015
9326697
fix it in PyPy
Sep 10, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion python/pyspark/cloudpickle.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -350,6 +350,11 @@ def save_global(self, obj, name=None, pack=struct.pack):
if new_override:
d['__new__'] = obj.__new__

# workaround for namedtuple (hijacked by PySpark)
if getattr(obj, '_is_namedtuple_', False):
self.save_reduce(_load_namedtuple, (obj.__name__, obj._fields))
return

self.save(_load_class)
self.save_reduce(typ, (obj.__name__, obj.__bases__, {"__doc__": obj.__doc__}), obj=obj)
d.pop('__doc__', None)
Expand Down Expand Up @@ -382,7 +387,7 @@ def save_instancemethod(self, obj):
self.save_reduce(types.MethodType, (obj.__func__, obj.__self__), obj=obj)
else:
self.save_reduce(types.MethodType, (obj.__func__, obj.__self__, obj.__self__.__class__),
obj=obj)
obj=obj)
dispatch[types.MethodType] = save_instancemethod

def save_inst(self, obj):
Expand Down Expand Up @@ -744,6 +749,14 @@ def _load_class(cls, d):
return cls


def _load_namedtuple(name, fields):
"""
Loads a class generated by namedtuple
"""
from collections import namedtuple
return namedtuple(name, fields)


"""Constructors for 3rd party libraries
Note: These can never be renamed due to client compatibility issues"""

Expand Down
1 change: 1 addition & 0 deletions python/pyspark/serializers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -359,6 +359,7 @@ def _hack_namedtuple(cls):
def __reduce__(self):
return (_restore, (name, fields, tuple(self)))
cls.__reduce__ = __reduce__
cls._is_namedtuple_ = True
return cls


Expand Down
5 changes: 5 additions & 0 deletions python/pyspark/tests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -218,6 +218,11 @@ def test_namedtuple(self):
p2 = loads(dumps(p1, 2))
self.assertEqual(p1, p2)

from pyspark.cloudpickle import dumps
P2 = loads(dumps(P))
p3 = P2(1, 3)
self.assertEqual(p1, p3)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, interesting: presumably P and P2 are different classes but instances created from them are still comparable for equality. Do we also need to check that those instances claim to belong to the same class? It seems way less likely that users could rely on the class comparison behavior, so probably not a huge priority to look at.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These instances should become to difference classes.


def test_itemgetter(self):
from operator import itemgetter
ser = CloudPickleSerializer()
Expand Down