python - cPickle.PicklingError : Could not serialize object: NotImplementedError

Question

在没有修改的情况下运行 Elephas 示例时出错:(即使使用 git 版本 pip install --no-cache-dir git+git://github.com/maxpumperla/elephas.git@master 也会出现该错误)

我用过的例子:https://github.com/maxpumperla/elephas/blob/master/examples/ml_pipeline_otto.py

(我试图启用 tf.compat.v1.enable_eager_execution() 但其他代码不适用于该设置)

pyspark_1      | 19/10/25 10:23:03 INFO SparkContext: Created broadcast 12 from broadcast at NativeMethodAccessorImpl.java:0
pyspark_1      | Traceback (most recent call last):
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/serializers.py", line 590, in dumps
pyspark_1      |     return cloudpickle.dumps(obj, 2)
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/cloudpickle.py", line 863, in dumps
pyspark_1      |     cp.dump(obj)
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/cloudpickle.py", line 260, in dump
pyspark_1      |     return Pickler.dump(self, obj)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 224, in dump
pyspark_1      |     self.save(obj)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 286, in save
pyspark_1      |     f(self, obj) # Call unbound method with explicit self
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 568, in save_tuple
pyspark_1      |     save(element)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 286, in save
pyspark_1      |     f(self, obj) # Call unbound method with explicit self
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/cloudpickle.py", line 406, in save_function
pyspark_1      |     self.save_function_tuple(obj)
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/cloudpickle.py", line 549, in save_function_tuple
pyspark_1      |     save(state)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 286, in save
pyspark_1      |     f(self, obj) # Call unbound method with explicit self
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 655, in save_dict
pyspark_1      |     self._batch_setitems(obj.iteritems())
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 687, in _batch_setitems
pyspark_1      |     save(v)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 286, in save
pyspark_1      |     f(self, obj) # Call unbound method with explicit self
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 606, in save_list
pyspark_1      |     self._batch_appends(iter(obj))
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 642, in _batch_appends
pyspark_1      |     save(tmp[0])
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 286, in save
pyspark_1      |     f(self, obj) # Call unbound method with explicit self
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/cloudpickle.py", line 660, in save_instancemethod
pyspark_1      |     obj=obj)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 401, in save_reduce
pyspark_1      |     save(args)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 286, in save
pyspark_1      |     f(self, obj) # Call unbound method with explicit self
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 554, in save_tuple
pyspark_1      |     save(element)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 331, in save
pyspark_1      |     self.save_reduce(obj=obj, *rv)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 425, in save_reduce
pyspark_1      |     save(state)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 286, in save
pyspark_1      |     f(self, obj) # Call unbound method with explicit self
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 655, in save_dict
pyspark_1      |     self._batch_setitems(obj.iteritems())
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 687, in _batch_setitems
pyspark_1      |     save(v)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 286, in save
pyspark_1      |     f(self, obj) # Call unbound method with explicit self
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 606, in save_list
pyspark_1      |     self._batch_appends(iter(obj))
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 642, in _batch_appends
pyspark_1      |     save(tmp[0])
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 331, in save
pyspark_1      |     self.save_reduce(obj=obj, *rv)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 425, in save_reduce
pyspark_1      |     save(state)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 286, in save
pyspark_1      |     f(self, obj) # Call unbound method with explicit self
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 655, in save_dict
pyspark_1      |     self._batch_setitems(obj.iteritems())
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 687, in _batch_setitems
pyspark_1      |     save(v)
pyspark_1      |   File "/usr/lib/python2.7/pickle.py", line 306, in save
pyspark_1      |     rv = reduce(self.proto)
pyspark_1      |   File "/usr/local/lib/python2.7/dist-packages/tensorflow/python/ops/resource_variable_ops.py", line 1152, in __reduce__
pyspark_1      |     initial_value=self.numpy(),
pyspark_1      |   File "/usr/local/lib/python2.7/dist-packages/tensorflow/python/ops/resource_variable_ops.py", line 906, in numpy
pyspark_1      |     "numpy() is only available when eager execution is enabled.")
pyspark_1      | NotImplementedError: numpy() is only available when eager execution is enabled.
pyspark_1      | Traceback (most recent call last):
pyspark_1      |   File "/home/ubuntu/./spark.py", line 169, in <module>
pyspark_1      |     fitted_pipeline = pipeline.fit(train_df)
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/ml/base.py", line 132, in fit
pyspark_1      |     return self._fit(dataset)
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/ml/pipeline.py", line 109, in _fit
pyspark_1      |     model = stage.fit(dataset)
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/ml/base.py", line 132, in fit
pyspark_1      |     return self._fit(dataset)
pyspark_1      |   File "/usr/local/lib/python2.7/dist-packages/elephas/ml_model.py", line 92, in _fit
pyspark_1      |     validation_split=self.get_validation_split())
pyspark_1      |   File "/usr/local/lib/python2.7/dist-packages/elephas/spark_model.py", line 151, in fit
pyspark_1      |     self._fit(rdd, epochs, batch_size, verbose, validation_split)
pyspark_1      |   File "/usr/local/lib/python2.7/dist-packages/elephas/spark_model.py", line 188, in _fit
pyspark_1      |     gradients = rdd.mapPartitions(worker.train).collect()
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/rdd.py", line 816, in collect
pyspark_1      |     sock_info = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/rdd.py", line 2532, in _jrdd
pyspark_1      |     self._jrdd_deserializer, profiler)
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/rdd.py", line 2434, in _wrap_function
pyspark_1      |     pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command)
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/rdd.py", line 2420, in _prepare_for_python_RDD
pyspark_1      |     pickled_command = ser.dumps(command)
pyspark_1      |   File "/home/ubuntu/spark-2.4.4-bin-hadoop2.7/python/pyspark/serializers.py", line 600, in dumps
pyspark_1      |     raise pickle.PicklingError(msg)
pyspark_1      | cPickle.PicklingError: Could not serialize object: NotImplementedError: numpy() is only available when eager execution is enabled.

Answer 1

问题似乎围绕着 spark_model.py 的 _fit 中 RDD 和 SparkWorker-s 的使用，关于 this line在交换到 TF 的 resource_variable_ops.py 之前:

gradients = rdd.mapPartitions(worker.train).collect()

无论是每个多线程还是使用其他抽象数据结构，TF 运行时都是拦截的，TF 认为它在 Eager 中并调用 Eager 方法 (.numpy()) ，但它不是 - 因此是错误。我非常怀疑对此是否存在“外部”解决方法，但以下对 TF 源的编辑可以解决问题(如下)。

基本上，它的工作方式是强制执行几乎所有可能的急切和非急切操作组合，以在图模式内外评估张量。

---

让我知道它是否有效。

---

# "/usr/local/lib/python2.7/dist-packages/tensorflow/python/ops/resource_variable_ops.py"
# line 1152
def __reduce__(self):
    # The implementation mirrors that of __deepcopy__.
    def K_eval(x, K):
        try:
            return K.get_value(K.to_dense(x))
        except:
            try:
                eval_fn = K.function([], [x])
                return eval_fn([])[0]
            except:
                return K.eval(x)
    try:
        import keras.backend as K
        initial_value = K_eval(self, K)
    except:
        import tensorflow.keras.backend as K
        initial_value = K_eval(self, K)

    return functools.partial(
        ResourceVariable,
        initial_value=initial_value,
        trainable=self.trainable,
        name=self._shared_name,
        dtype=self.dtype,
        constraint=self.constraint,
        distribute_strategy=self._distribute_strategy), ()