Spark olmayan bir ortamda bir pyspark ML modeli yükleme

Question

Python'da bir makine öğrenimi modelinin dağıtımıyla ilgileniyorum, bu nedenle bir sunucuya yapılan isteklerle tahminler yapılabilir.

Bir Cloudera kümesi oluşturacağım ve kütüphaneyi pyspark kullanarak, modelleri geliştirmek için Spark'den faydalanacağım. Modelin sunucuda çalıştırılması için nasıl kaydedilebileceğini bilmek isterim.

Farklı algoritmaların .save işlevlerine sahip olduğunu gördüm (bu iletide How to save and load MLLib model in Apache Spark yanıtı var), ancak sunucu Spark olmadan ve Cloudera kümesinde değil, farklı bir makinede olacağından .load ve .predict işlevlerini kullanmak mümkün olup olmadığını bilmek.

Spark altında olmaksızın tahmin için pyspark kütüphane işlevlerini kullanarak yapılabilir mi? Ya da modeli kaydetmek ve başka bir yerde kullanmak için herhangi bir dönüşüm yapmak zorunda mıyım?

Answer 1

Bu, tam çözüm olmayabilir.

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 yapabileceğiniz böyle

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

şey. ref: https://loads.pickle.me.uk/2016/04/04/deploying-a-scikit-learn-classifier-to-production/ kıvılcım

: Bu optimize edilmemiş olabilir http://spark.apache.org/docs/latest/mllib-collaborative-filtering.html

Answer 2

ben bu çalışma kodu var bir saat geçirdikten sonra

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

Bu kod yaratacak Bir küme küme modeli ve dosya sistemi içine kaydedin.

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

Üstü şişeye yazılmış benim DİNLENME API olduğunu.

Aramayı http://127.0.0.1:5000/ yapın. Yanıtı tarayıcıda görebilirsiniz.

Answer 3

MLeap'a (katkıda bulunduğum bir projeye) göz atın - tüm ML Pipeline'lerinin (yalnızca tahmin edicinin değil) serileştirme/de-serileştirme ve kıvılcım bağlamına dayanmayan bir yürütme motoru, dağıtılmış veri sağlar. çerçeveler ve yürütme planları.

Bugün itibarıyla, MLeap'ın modelleri yürütme zamanı python bağlamaları, yalnızca scala/java içermiyor, ancak bunları eklemenin zor olması gerekmiyor. Spark eğitimli boru hatlarınız ve modellerinizden bir puanlama motoru oluşturma konusunda yardıma ihtiyacınız varsa, kendime ve diğer MLeap geliştiricilerine github'a ulaşmaktan çekinmeyin.