在 Spark\PySpark 中保存\加载模型的正确方法是什么答案

【问题标题】：What is the right way to save\load models in Spark\PySpark在 Spark\PySpark 中保存\加载模型的正确方法是什么
【发布时间】：2015-05-29 02:14:07
【问题描述】：

我正在使用 PySpark 和 MLlib 使用 Spark 1.3.0，我需要保存和加载我的模型。我使用这样的代码（取自官方documentation）

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，但遇到了与此处所述相同的问题 Save Apache Spark mllib model in python

【问题讨论】：

标签： python apache-spark pyspark apache-spark-mllib

【解决方案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】：

this pull request 于 2015 年 3 月 28 日合并（您的问题最后一次编辑的第二天），此问题已得到解决。

您只需从 GitHub (git clone git://github.com/apache/spark.git -b branch-1.3) 克隆/获取最新版本，然后使用 $ mvn -DskipTests clean package 构建它（按照 spark/README.md 中的说明）。

注意：我在构建 Spark 时遇到了麻烦，因为 Maven 不稳定。我通过使用 $ update-alternatives --config mvn 并选择优先级为 150 的“路径”解决了这个问题，无论这意味着什么。 Explanation here.

【讨论】：

是的，我看过这个 PR，谢谢！但我还没有尝试自己构建 Spark。还要感谢 Maven 的提示 :)

【解决方案3】：

我也遇到了这个问题——它看起来像一个错误。我已经向spark jira举报了。

【讨论】：

【解决方案4】：

在 ML 中使用 pipeline 训练模型，然后使用 MLWriter 和 MLReader 保存模型并读回。

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

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

【讨论】：

谢谢，但这个问题已经很老了 :) 自从提出这个问题以来，很多事情都发生了变化。