在非 Spark 环境中加载 pyspark ML 模型

Posted 2023-04-15

【中文标题】在非 Spark 环境中加载 pyspark ML 模型

【英文标题】：Loading a pyspark ML model in a non-Spark environment

【发布时间】：2016-11-21 08:11:49

【问题描述】：

我有兴趣在 python 中部署机器学习模型，因此可以通过对服务器的请求进行预测。

我将创建一个 Cloudera 集群，并通过使用库 pyspark 来利用 Spark 开发模型。我想知道如何保存模型以便在服务器上使用它。

我已经看到不同的算法具有 .save 功能（就像在这篇文章 How to save and load MLLib model in Apache Spark 中回答的那样），但是由于服务器将在没有 Spark 而不是在 Cloudera 集群中的另一台机器上，所以我没有不知道是否可以使用他们的 .load 和 .predict 函数。

是否可以使用 pyspark 库函数在没有 Spark 的情况下进行预测？或者我是否必须进行任何转换才能保存模型并在其他地方使用它？

【问题讨论】：

我相信你需要有火花。我要说的一件事是，您可以在 python 中创建一个简单的 rest api 并加载模型文件并发送响应。

我添加了一个有效的答案

【参考方案1】：

花了一个小时后我得到了这个工作代码，这可能没有优化，

Mymodel.py：

import os
import sys

# Path for spark source folder
os.environ['SPARK_HOME']="E:\\Work\\spark\\installtion\\spark"

# Append pyspark  to Python Path
sys.path.append("E:\\Work\\spark\\installtion\\spark\\python")

try:
    from pyspark.ml.feature import StringIndexer
    # $example on$
    from numpy import array
    from math import sqrt
    from pyspark import SparkConf
    # $example off$

    from pyspark import SparkContext
    # $example on$
    from pyspark.mllib.clustering import KMeans, KMeansModel

    print ("Successfully imported Spark Modules")

except ImportError as e:
    sys.exit(1)


if __name__ == "__main__":
    sconf = SparkConf().setAppName("KMeansExample").set('spark.sql.warehouse.dir', 'file:///E:/Work/spark/installtion/spark/spark-warehouse/')
    sc = SparkContext(conf=sconf)  # SparkContext
    parsedData =  array([0.0,0.0, 1.0,1.0, 9.0,8.0, 8.0,9.0]).reshape(4,2)
    clusters = KMeans.train(sc.parallelize(parsedData), 2, maxIterations=10,
                            runs=10, initializationMode="random")
    clusters.save(sc, "mymodel")  // this will save model to file system
    sc.stop()

此代码将创建一个 kmean 集群模型并将其保存在文件系统中。

API.py

from flask import jsonify, request, Flask
from sklearn.externals import joblib
import os
import sys

# Path for spark source folder
os.environ['SPARK_HOME']="E:\\Work\\spark\\installtion\\spark"

# Append pyspark  to Python Path
sys.path.append("E:\\Work\\spark\\installtion\\spark\\python")

try:
    from pyspark.ml.feature import StringIndexer
    # $example on$
    from numpy import array
    from math import sqrt
    from pyspark import SparkConf
    # $example off$

    from pyspark import SparkContext
    # $example on$
    from pyspark.mllib.clustering import KMeans, KMeansModel

    print ("Successfully imported Spark Modules")

except ImportError as e:
    sys.exit(1)


app = Flask(__name__)

@app.route('/', methods=['GET'])
def predict():

    sconf = SparkConf().setAppName("KMeansExample").set('spark.sql.warehouse.dir', 'file:///E:/Work/spark/installtion/spark/spark-warehouse/')
    sc = SparkContext(conf=sconf)  # SparkContext
    sameModel = KMeansModel.load(sc, "clus")  // load from file system 

    response = sameModel.predict(array([0.0, 0.0]))  // pass your data

    return jsonify(response)

if __name__ == '__main__':
    app.run()

上面是我用烧瓶写的 REST api。

拨打http://127.0.0.1:5000/。您可以在浏览器中看到响应。

【讨论】：

我想知道调用 clusters.save 时模型以哪种格式保存。提前致谢。

您好，谢谢您的回答。但有一件事我不确定。我可以在只安装了 python 的机器上运行 API.py 脚本吗？还是我还需要安装 Spark？那样的话，安装单机版就够了吗？

@MarcialGonzalez，是的，我们必须在服务器上安装 Spark，或者你可以做另一件事，在你的 rest 和 spark ml 服务器之间建立一个基于端口的通信。

@MarcialGonzalez，在我的生产中，我们有一个 REST api 暴露给客户端，我们的 ML 服务器正在运行，REST api 将使用基于端口与 ML 服务器通信并返回响应

是否可以通过使用 pickle 或 joblib 来持久化 pyspark 生成的 spark ml 模型？想法是将其导出并加载到仅安装 Python 的机器中。

【参考方案2】：

看看MLeap（我参与的一个项目）——它提供了整个 ML 管道（不仅仅是估计器）的序列化/反序列化和一个不依赖于 spark 上下文的分布式执行引擎数据帧和执行计划。

截至今天，MLeap 用于执行模型的运行时没有 python 绑定，只有 scala/java，但添加它们应该不会很复杂。如果您需要帮助从 Spark 训练的管道和模型创建评分引擎，请随时在 github 上与我自己和其他 MLeap 开发人员联系。

【讨论】：

我刚刚创建了***.com/questions/tagged/mleap 标签，您可能想关注它。是否有计划将 mleap 集成到主 Spark 项目/分支中？对 Java 8 的支持如何？

【参考方案3】：

这可能不是完整的解决方案。

模型.py

from sklearn.externals import joblib
from sklearn.pipeline import make_pipeline
from sklearn.feature_extraction.text import HashingVectorizer
from sklearn.svm import LinearSVC

# code to load training data into X_train, y_train, split train/test set

vec = HashingVectorizer()
svc = LinearSVC()
clf = make_pipeline(vec, svc)
svc.fit(X_train, y_train)

joblib.dump('class1': clf, 'models', compress=9)

myRest.py

from flask import jsonify, request, Flask
from sklearn.externals import joblib

models = joblib.load('models')
app = Flask(__name__)

@app.route('/', methods=['POST'])
def predict():
    text = request.form.get('text')
    results = 
    for name, clf in models.iteritems():
        results[name] = clf.predict([text])[0]
    return jsonify(results)

if __name__ == '__main__':
    app.run()

你可以做这样的事情。参考：https://loads.pickle.me.uk/2016/04/04/deploying-a-scikit-learn-classifier-to-production/

对于火花：http://spark.apache.org/docs/latest/mllib-collaborative-filtering.html

【讨论】：

恐怕根本不是解决办法。 PySpark ml 不是 scikit-learn。

@LostInOverflow，我也知道我为 scikit-learn 添加了示例。确实我接受你的评论。但是我们甚至可以像这样加载 spark ml 模型。 sameModel = MatrixFactorizationModel.load(sc, "target/tmp/myCollaborativeFilter") 。检查此链接：spark.apache.org/docs/latest/mllib-collaborative-filtering.html

可以，但至少需要本地模式“集群”。所以它不是非 Spark 环境。

@LostInOverflow 是的，并使用工作样本检查我的另一个答案