Каков правильный способ сохранения\загрузки моделей в Spark\PySpark

Я работаю с Spark 1.3.0 с помощью PySpark и MLlib, и мне нужно сохранять и загружать мои модели. Я использую такой код (взято из официальной документации)

from pyspark.mllib.recommendation import ALS, MatrixFactorizationModel, Rating

data = sc.textFile("data/mllib/als/test.data")
ratings = data.map(lambda l: l.split(',')).map(lambda l: Rating(int(l[0]), int(l[1]), float(l[2])))
rank = 10
numIterations = 20
model = ALS.train(ratings, rank, numIterations)
testdata = ratings.map(lambda p: (p[0], p[1]))
predictions = model.predictAll(testdata).map(lambda r: ((r[0], r[1]), r[2]))
predictions.collect() # shows me some predictions
model.save(sc, "model0")

# Trying to load saved model and work with it
model0 = MatrixFactorizationModel.load(sc, "model0")
predictions0 = model0.predictAll(testdata).map(lambda r: ((r[0], r[1]), r[2]))

После того, как я попытаюсь использовать model0, я получаю длинную трассировку, которая заканчивается следующим:

Py4JError: An error occurred while calling o70.predict. Trace:
py4j.Py4JException: Method predict([class org.apache.spark.api.java.JavaRDD]) does not exist
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:333)
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:342)
    at py4j.Gateway.invoke(Gateway.java:252)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:207)
    at java.lang.Thread.run(Thread.java:745)

Итак, мой вопрос: я что-то делаю неправильно? Насколько я отлаживал свои модели, хранятся (локально и на HDFS), и они содержат много файлов с некоторыми данными. У меня такое чувство, что модели сохранены правильно, но, вероятно, они загружены неправильно. Я также googled вокруг, но не нашел ничего связанного.

Похоже, эта функция save\load была добавлена недавно в Spark 1.3.0, и из-за этого у меня есть еще один вопрос - каков был рекомендованный способ сохранить\загружать модели до версии 1.3.0? Я не нашел хороших способов сделать это, по крайней мере для Python. Я также попытался Pickle, но столкнулся с теми же проблемами, что описаны здесь Сохранить Apache Spark mllib model в python

Ответ 1

Один способ сохранить модель (в Scala, но, вероятно, похож на Python):

// persist model to HDFS
sc.parallelize(Seq(model), 1).saveAsObjectFile("linReg.model")

Сохраненная модель затем может быть загружена как:

val linRegModel = sc.objectFile[LinearRegressionModel]("linReg.model").first()

См. также question

Подробнее см. (ref)

Ответ 2

По состоянию этот запрос на перенос объединился 28 марта 2015 года (на следующий день после последнего изменения вашего вопроса) эта проблема была решена.

Вам просто нужно клонировать/извлекать последнюю версию из GitHub (git clone git://github.com/apache/spark.git -b branch-1.3), а затем строить ее (следуя инструкциям в spark/README.md) с помощью $ mvn -DskipTests clean package.

Примечание. Я столкнулся с проблемой создания Искры, потому что Мейвен был неуклюжим. Я решил эту проблему, используя $ update-alternatives --config mvn и выбрав "путь", который имел приоритет: 150, что бы это ни значило. Объяснение здесь.

Ответ 3

Я тоже сталкиваюсь с этим - это похоже на ошибку. Я сообщил spark jira.

Ответ 4

Используйте конвейер в ML для обучения модели, а затем используйте MLWriter и MLReader, чтобы сохранить модели и прочитать их.

from pyspark.ml import Pipeline
from pyspark.ml import PipelineModel

pipeTrain.write().overwrite().save(outpath)
model_in = PipelineModel.load(outpath)