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

Question

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

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

我已经看到不同的算法具有.save函数(就像在本文中如何在Apache Spark中保存和加载MLLib模型一样),但是因为服务器将在没有Spark的不同机器中而不在Cloudera集群中,我不知道是否可以使用他们的.load和.predict函数.

可以通过使用pyspark库函数进行预测而不使用Spark吗？或者我是否必须进行任何转换才能保存模型并在其他地方使用它？

Answer 1

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

我的模型.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()

上面是我用 Flask 编写的 REST api。

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