Fixed named tuple issue

python/pyspark/cloudpickle.py

@@ -547,6 +547,10 @@ def save_global(self, obj, name=None, pack=struct.pack):
             if new_override:
                 d['__new__'] = obj.__new__
 
+            # namedtuple is a special case for Spark
+            if getattr(obj, '_is_namedtuple_', False):
+                self.save_reduce(_load_namedtuple, (obj.__name__, obj._fields))
+                return
             self.save_reduce(typ, (obj.__name__, obj.__bases__, d), obj=obj)
         else:
             raise pickle.PicklingError("Can't pickle %r" % obj)
@@ -1000,6 +1004,13 @@ def _find_module(mod_name):
             file.close()
     return path, description
 
+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"""