我正在尝试在群集上分发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)```
另外,该模型实际上是创建的,我可以做print(spark_model)
并将得到这个<elephas.spark_model.SparkModel object at 0x7efce0abfcd0>
。发生错误spark_model.train
。
我已经使用安装了elephas pip2 install git+https://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 Notebook中使用。两者都导致相同的问题。
谁能给我任何指示?这是与elephas相关还是PySpark问题?
编辑:我还上传了虚拟环境的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
从上面得到错误。
我找到了关于如何正确地将虚拟环境加载到主服务器和所有从属计算机的解决方案:
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
本文收集自互联网,转载请注明来源。
如有侵权,请联系 [email protected] 删除。
我来说两句