Skip to content

replace vendored cloudpickle in pyspark for python 3.7 support? #305

Closed
Closed
replace vendored cloudpickle in pyspark for python 3.7 support?#305

Description

@AlJohri
AlJohri
opened on Sep 8, 2019

I tried to replace the default PySpark (python) serializer doing the following:

import cloudpickle
from pyspark.serializers import PickleSerializer
print(cloudpickle.__file__) # returns my newly installed cloudpickle

class CloudPickleSerializer(PickleSerializer):

    def dumps(self, obj):
        print('serializing...')
        print(cloudpickle.__file__)
        return cloudpickle.dumps(obj, 2)

sc = SparkContext(serializer=CloudPickleSerializer)
spark = SparkSession(sc)

However, my custom CloudPickleSerializer never gets called. As you can see in the stacktrace below its still using the vendored cloudpickle.

Looking at the pyspark.cloudpickle module, it doesn't have this feature #189. It will be released in 3.0.0 according to JIRA apache/spark@a67e842
https://issues.apache.org/jira/browse/SPARK-27000

Is there either something wrong with the way in which I'm trying to set the pyspark serializer or is there another way to replace the cloudpickle version vendored with pyspark.

Traceback (most recent call last):
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 590, in dumps
    return cloudpickle.dumps(obj, 2)
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 863, in dumps
    cp.dump(obj)
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 260, in dump
    return Pickler.dump(self, obj)
  File "/opt/python3/lib/python3.7/pickle.py", line 437, in dump
    self.save(obj)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 771, in save_tuple
    save(element)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 400, in save_function
    self.save_function_tuple(obj)
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 549, in save_function_tuple
    save(state)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 856, in save_dict
    self._batch_setitems(obj.items())
  File "/opt/python3/lib/python3.7/pickle.py", line 882, in _batch_setitems
    save(v)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 816, in save_list
    self._batch_appends(obj)
  File "/opt/python3/lib/python3.7/pickle.py", line 840, in _batch_appends
    save(x)
  File "/opt/python3/lib/python3.7/pickle.py", line 549, in save
    self.save_reduce(obj=obj, *rv)
  File "/opt/python3/lib/python3.7/pickle.py", line 662, in save_reduce
    save(state)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 856, in save_dict
    self._batch_setitems(obj.items())
  File "/opt/python3/lib/python3.7/pickle.py", line 882, in _batch_setitems
    save(v)
  File "/opt/python3/lib/python3.7/pickle.py", line 549, in save
    self.save_reduce(obj=obj, *rv)
  File "/opt/python3/lib/python3.7/pickle.py", line 662, in save_reduce
    save(state)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 856, in save_dict
    self._batch_setitems(obj.items())
  File "/opt/python3/lib/python3.7/pickle.py", line 882, in _batch_setitems
    save(v)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 816, in save_list
    self._batch_appends(obj)
  File "/opt/python3/lib/python3.7/pickle.py", line 840, in _batch_appends
    save(x)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 771, in save_tuple
    save(element)
  File "/opt/python3/lib/python3.7/pickle.py", line 549, in save
    self.save_reduce(obj=obj, *rv)
  File "/opt/python3/lib/python3.7/pickle.py", line 662, in save_reduce
    save(state)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 856, in save_dict
    self._batch_setitems(obj.items())
  File "/opt/python3/lib/python3.7/pickle.py", line 882, in _batch_setitems
    save(v)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 816, in save_list
    self._batch_appends(obj)
  File "/opt/python3/lib/python3.7/pickle.py", line 840, in _batch_appends
    save(x)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 771, in save_tuple
    save(element)
  File "/opt/python3/lib/python3.7/pickle.py", line 549, in save
    self.save_reduce(obj=obj, *rv)
  File "/opt/python3/lib/python3.7/pickle.py", line 633, in save_reduce
    save(cls)
  File "/opt/python3/lib/python3.7/pickle.py", line 518, in save
    self.save_global(obj)
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 632, in save_global
    return self.save_dynamic_class(obj)
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 491, in save_dynamic_class
    save(clsdict)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 856, in save_dict
    self._batch_setitems(obj.items())
  File "/opt/python3/lib/python3.7/pickle.py", line 882, in _batch_setitems
    save(v)
  File "/opt/python3/lib/python3.7/pickle.py", line 524, in save
    rv = reduce(self.proto)
TypeError: can't pickle _abc_data objects
Traceback (most recent call last):
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 590, in dumps
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 863, in dumps
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 260, in dump
  File "/opt/python3/lib/python3.7/pickle.py", line 437, in dump
    self.save(obj)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 771, in save_tuple
    save(element)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 400, in save_function
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 549, in save_function_tuple
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 856, in save_dict
    self._batch_setitems(obj.items())
  File "/opt/python3/lib/python3.7/pickle.py", line 882, in _batch_setitems
    save(v)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 816, in save_list
    self._batch_appends(obj)
  File "/opt/python3/lib/python3.7/pickle.py", line 840, in _batch_appends
    save(x)
  File "/opt/python3/lib/python3.7/pickle.py", line 549, in save
    self.save_reduce(obj=obj, *rv)
  File "/opt/python3/lib/python3.7/pickle.py", line 662, in save_reduce
    save(state)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 856, in save_dict
    self._batch_setitems(obj.items())
  File "/opt/python3/lib/python3.7/pickle.py", line 882, in _batch_setitems
    save(v)
  File "/opt/python3/lib/python3.7/pickle.py", line 549, in save
    self.save_reduce(obj=obj, *rv)
  File "/opt/python3/lib/python3.7/pickle.py", line 662, in save_reduce
    save(state)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 856, in save_dict
    self._batch_setitems(obj.items())
  File "/opt/python3/lib/python3.7/pickle.py", line 882, in _batch_setitems
    save(v)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 816, in save_list
    self._batch_appends(obj)
  File "/opt/python3/lib/python3.7/pickle.py", line 840, in _batch_appends
    save(x)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 771, in save_tuple
    save(element)
  File "/opt/python3/lib/python3.7/pickle.py", line 549, in save
    self.save_reduce(obj=obj, *rv)
  File "/opt/python3/lib/python3.7/pickle.py", line 662, in save_reduce
    save(state)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 856, in save_dict
    self._batch_setitems(obj.items())
  File "/opt/python3/lib/python3.7/pickle.py", line 882, in _batch_setitems
    save(v)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 816, in save_list
    self._batch_appends(obj)
  File "/opt/python3/lib/python3.7/pickle.py", line 840, in _batch_appends
    save(x)
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 771, in save_tuple
    save(element)
  File "/opt/python3/lib/python3.7/pickle.py", line 549, in save
    self.save_reduce(obj=obj, *rv)
  File "/opt/python3/lib/python3.7/pickle.py", line 633, in save_reduce
    save(cls)
  File "/opt/python3/lib/python3.7/pickle.py", line 518, in save
    self.save_global(obj)
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 632, in save_global
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 491, in save_dynamic_class
  File "/opt/python3/lib/python3.7/pickle.py", line 504, in save
    f(self, obj) # Call unbound method with explicit self
  File "/opt/python3/lib/python3.7/pickle.py", line 856, in save_dict
    self._batch_setitems(obj.items())
  File "/opt/python3/lib/python3.7/pickle.py", line 882, in _batch_setitems
    save(v)
  File "/opt/python3/lib/python3.7/pickle.py", line 524, in save
    rv = reduce(self.proto)
TypeError: can't pickle _abc_data objects

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/mnt/repo/predictor/spark/main.py", line 643, in <module>
    H.run_days(main, dts[0], dts[-1], print_str=None)(spark_session, arg)
  File "/mnt/repo/predictor/spark/helper.py", line 74, in wrapper
    result = func(*args, **kwargs)
  File "/mnt/repo/predictor/spark/helper.py", line 18, in decorator
    result = func(*args, **kw)
  File "/mnt/repo/predictor/spark/main.py", line 621, in main
    df_predict = wlogger(predict)(df_sum)
  File "/mnt/repo/predictor/spark/helper.py", line 18, in decorator
    result = func(*args, **kw)
  File "/mnt/repo/predictor/spark/main.py", line 521, in predict
    return df.withColumn("prediction", predict_inner(*columns))
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/udf.py", line 189, in wrapper
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/udf.py", line 167, in __call__
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/udf.py", line 151, in _judf
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/udf.py", line 160, in _create_judf
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/udf.py", line 35, in _wrap_function
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2420, in _prepare_for_python_RDD
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 600, in dumps
_pickle.PicklingError: Could not serialize object: TypeError: can't pickle _abc_data objects

Activity

  1. robertnishihara commented on Sep 8, 2019

    @robertnishihara
    robertnishihara
    on Sep 8, 2019

    Just a wild guess, but try doing pip uninstall typing everywhere.

  2. AlJohri commented on Sep 8, 2019

    @AlJohri
    AlJohri
    on Sep 8, 2019
    Author

    I actually got around it by doing this monkeypatch:

    import cloudpickle
import pyspark.serializers
pyspark.serializers.cloudpickle = cloudpickle

  3. ogrisel commented on Sep 11, 2019

    @ogrisel
    ogrisel
    on Sep 11, 2019
    Contributor

    Is there anything left to do on the cloudpickle project side? If not, we can close this issue.

    Note that cloudpickle 1.2.2 was released yesterday.

  4. AlJohri commented on Sep 11, 2019

    @AlJohri
    AlJohri
    on Sep 11, 2019
    Author

    Potentially worth documenting this somewhere? I’ll close the issue for now- thanks!

  5. AlJohri
    closed this as completedon Sep 11, 2019
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions