Elephave未加载在PySpark：没有名为elephas.spark模块_model

Question

我正试图在集群上分发Keras培训，并使用Elephas来实现这一点。但是，当运行来自Elephas文档的基本示例时(https://github.com/maxpumperla/elephas):

from elephas.utils.rdd_utils import to_simple_rdd
rdd = to_simple_rdd(sc, x_train, y_train)
from elephas.spark_model import SparkModel
from elephas import optimizers as elephas_optimizers
sgd = elephas_optimizers.SGD()
spark_model = SparkModel(sc, model, optimizer=sgd, frequency='epoch', mode='asynchronous', num_workers=2)
spark_model.train(rdd, nb_epoch=epochs, batch_size=batch_size, verbose=1, validation_split=0.1)

我得到以下错误：

 ImportError: No module named elephas.spark_model



```Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 5.0 failed 4 times, most recent failure: Lost task 1.3 in stage 5.0 (TID 58, xxxx, executor 8): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/xx/xx/hadoop/yarn/local/usercache/xx/appcache/application_151xxx857247_19188/container_1512xxx247_19188_01_000009/pyspark.zip/pyspark/worker.py", line 163, in main
    func, profiler, deserializer, serializer = read_command(pickleSer, infile)
  File "/xx/xx/hadoop/yarn/local/usercache/xx/appcache/application_151xxx857247_19188/container_1512xxx247_19188_01_000009/pyspark.zip/pyspark/worker.py", line 54, in read_command
    command = serializer._read_with_length(file)
  File /yarn/local/usercache/xx/appcache/application_151xxx857247_19188/container_1512xxx247_19188_01_000009/pyspark.zip/pyspark/serializers.py", line 169, in _read_with_length
    return self.loads(obj)
  File "/yarn//local/usercache/xx/appcache/application_151xxx857247_19188/container_1512xxx247_19188_01_000009/pyspark.zip/pyspark/serializers.py", line 454, in loads
    return pickle.loads(obj)
ImportError: No module named elephas.spark_model

    at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193)
    at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:234)
    at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152)
    at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:63)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
    at org.apache.spark.scheduler.Task.run(Task.scala:99)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)```

此外，模型实际上已经创建，我可以执行打印（spark\u model）并将获得此

我已经安装了elephad使用pip2安装githttps://github.com/maxpumperla/elephas，也许这是相关的。

我使用PySpark 2.1.1、Keras 2.1.4和Python 2.7。我已尝试使用spark submit运行它：

PYSPARK_DRIVER_PYTHON=`which python` spark-submit --driver-memory 1G  filname.py

而且直接放在Jupyter笔记本上。两者都会导致相同的问题。

有人能给我指点吗？这与elephas有关还是Pypark问题？

编辑：我也上传虚拟环境的zip文件，并在脚本中调用它：

virtualenv spark_venv --relocatable
cd spark_venv 
zip -qr ../spark_venv.zip *

PYSPARK_DRIVER_PYTHON=`which python` spark-submit --driver-memory 1G --py-files spark_venv.zip filename.py

然后在文件中，我执行以下操作：

sc.addPyFile("spark_venv.zip")

在这个keras导入之后没有任何问题，但是我仍然从上面得到elephas错误。




Answer 1

您应该将elephave库作为参数添加到您的Spark-提交命令中。

引用官方指南：

对于Python，您可以使用Spark-提交的--py-file参数来添加. py、. zip或. php文件，以便与您的应用程序一起分发。如果您依赖于多个Python文件，我们建议将它们打包到. zip或. egg中。

官方指南

Answer 2

我找到了一个解决方案，可以正确地将虚拟环境加载到主工作环境和所有从工作环境：

virtualenv venv --relocatable
cd venv 
zip -qr ../venv.zip *

PYSPARK_PYTHON=./SP/bin/python spark-submit --master yarn --deploy-mode cluster --conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./SP/bin/python --driver-memory 4G --archives venv.zip#SP filename.py

GitHub问题的更多细节：https://github.com/maxpumperla/elephas/issues/80#issuecomment-371073492