Qual è il modo giusto per salvare \ caricare i modelli in Spark \ PySpark

Question

Sto lavorando con Spark 1.3.0 usando PySpark e MLlib e ho bisogno di salvare e caricare i miei modelli. Io uso il codice come questo (preso dal sito ufficiale 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])) 

Dopo provo ad usare model0 ho una lunga traceback, che termina con questo:

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) 

Quindi la mia domanda è - sono io fare qualcosa di sbagliato? Per quanto riguarda il debug dei miei modelli, sono memorizzati (localmente e su HDFS) e contengono molti file con alcuni dati. Ho la sensazione che i modelli vengano salvati correttamente, ma probabilmente non vengono caricati correttamente. Ho anche cercato su google, ma non ho trovato nulla di relativo.

Sembra che questa funzione di salvataggio \ caricamento sia stata aggiunta recentemente in Spark 1.3.0 e per questo ho un'altra domanda: quale era il modo consigliato di salvare \ caricare i modelli prima della versione 1.3.0? Non ho trovato nessun modo per farlo, almeno per Python. Ho anche provato Pickle, ma ho affrontato gli stessi problemi descritti qui Save Apache Spark mllib model in python

Answer 1

Mi sono imbattuto anche in questo - sembra un bug. Ho segnalato a spark jira.

Answer 2

A partire da this pull request unito il 28 marzo 2015 (un giorno dopo l'ultima modifica della tua domanda) questo problema è stato risolto.

Hai solo bisogno di clonare/recuperare l'ultima versione da GitHub (git clone git://github.com/apache/spark.git -b branch-1.3) e poi costruirlo (seguendo le istruzioni nel spark/README.md) con $ mvn -DskipTests clean package.

Nota: ho avuto problemi a costruire Spark perché Maven era impazzita. Ho risolto il problema utilizzando $ update-alternatives --config mvn e selezionando il "percorso" con priorità: 150, qualunque cosa significhi. Explanation here.

Answer 3

Un modo per risparmiare un modello (in Scala, ma probabilmente è simile in Python):

// persist model to HDFS 
sc.parallelize(Seq(model), 1).saveAsObjectFile("linReg.model") 

modello salvato può essere caricato come:

val linRegModel = sc.objectFile[LinearRegressionModel]("linReg.model").first() 

Vedi anche legati question

Per ulteriori dettagli vedere (ref)

Answer 4

Uso pi peline in ML per addestrare il modello, quindi utilizzare MLWriter e MLReader per salvare i modelli e leggerli di nuovo.

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

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