Загрузка модели pyspark ML в среде без искры

Я заинтересован в развертывании модели машинного обучения в python, поэтому прогнозы могут быть сделаны с помощью запросов к серверу.

Я создам кластер Cloudera и воспользуюсь Spark для разработки моделей с помощью библиотеки pyspark. Я хотел бы знать, как модель может быть сохранена, чтобы использовать ее на сервере.

Я видел, что разные алгоритмы имеют .save-функции (например, ответ на этот пост Как сохранить и загрузить модель MLLib в Apache Spark), но поскольку сервер будет находиться на другой машине без Spark, а не в кластере Cloudera, я не знаю, можно ли использовать их .load и .predict функции.

Можно ли это сделать, используя функции библиотеки pyspark для прогнозирования без Spark? Или мне придется делать какие-либо преобразования, чтобы сохранить модель и использовать ее в другом месте?

Ответ 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/. Вы можете увидеть ответ в браузере.

Ответ 2

Взгляните на MLeap (проект, в который я вношу) - он обеспечивает сериализацию/де-сериализацию целых ML-трубопроводов ( а не только оценщик) и механизм выполнения, который не полагается на контекст искры, распределенные кадры данных и планы выполнения.

На сегодняшний день среда исполнения MLeap для выполнения моделей не имеет привязок к python, только scala/java, но не должна быть сложной для их добавления. Не стесняйтесь обращаться к github к себе и другим разработчикам MLeap, если вам нужна помощь в создании двигателя подсчета очков из ваших подготовленных Spark трубопроводов и моделей.

Ответ 3

Это может быть не полное решение.

Model.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()

что-то вроде этого вы можете сделать. ref: 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