Как правильно сохранять \ загружать модели в 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) и содержат много файлов с некоторыми данными. Такое ощущение, что модели сохраняются правильно, но, вероятно, они не загружаются правильно. Я также погуглил, но не нашел ничего связанного.

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


python apache-spark pyspark apache-spark-mllib
person artemdevel    schedule 25.03.2015    source источник

Ответы (4)


arrow_upward
7
arrow_downward

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

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

После этого сохраненную модель можно загрузить как:

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

См. Также соответствующий вопрос

Подробнее см. (ссылка)

person Neil    schedule 19.09.2015

arrow_upward
5
arrow_downward

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

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

Примечание: у меня возникли проблемы при сборке Spark, потому что Maven шатался. Я решил эту проблему, используя $ update-alternatives --config mvn и выбрав «путь» с приоритетом 150, что бы это ни значило. Пояснения здесь.

person emmagras    schedule 31.03.2015
comment
Да, я видел этот пиар, спасибо! Но я еще не пробовал создавать Spark. Также спасибо за подсказку для Maven :) - person artemdevel; 31.03.2015

arrow_upward
2
arrow_downward

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

person Charles Hayden    schedule 27.03.2015

arrow_upward
1
arrow_downward

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

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

pipeTrain.write().overwrite().save(outpath)
model_in = PipelineModel.load(outpath)
person Rong Du    schedule 13.10.2017
comment
спасибо, но это очень старый вопрос :) многое изменилось с тех пор, как его задали. - person artemdevel; 13.10.2017