Wie kann ich LIBSVM-Modelle (gespeichert mit LIBSVM) in PySpark lesen?

Question

Ich habe ein LIBSVM-Skalierungsmodell (generiert mit SVM-Skala), das ich in PySpark portieren möchte. Ich habe versucht, ganz naiv die folgenden:

scaler_path = "path to model" 
a = MinMaxScaler().load(scaler_path) 

Aber ich einen Fehler geworfen, ein Metadaten-Verzeichnis erwartet:

Py4JJavaErrorTraceback (most recent call last) 
<ipython-input-22-1942e7522174> in <module>() 
----> 1 a = MinMaxScaler().load(scaler_path) 

/srv/data/spark/spark-2.0.0-bin-hadoop2.6/python/pyspark/ml/util.pyc in load(cls, path) 
    226  def load(cls, path): 
    227   """Reads an ML instance from the input path, a shortcut of `read().load(path)`.""" 
--> 228   return cls.read().load(path) 
    229 
    230 

/srv/data/spark/spark-2.0.0-bin-hadoop2.6/python/pyspark/ml/util.pyc in load(self, path) 
    174   if not isinstance(path, basestring): 
    175    raise TypeError("path should be a basestring, got type %s" % type(path)) 
--> 176   java_obj = self._jread.load(path) 
    177   if not hasattr(self._clazz, "_from_java"): 
    178    raise NotImplementedError("This Java ML type cannot be loaded into Python currently: %r" 

/usr/local/lib/python2.7/dist-packages/py4j/java_gateway.pyc in __call__(self, *args) 
    1131   answer = self.gateway_client.send_command(command) 
    1132   return_value = get_return_value(
-> 1133    answer, self.gateway_client, self.target_id, self.name) 
    1134 
    1135   for temp_arg in temp_args: 

/srv/data/spark/spark-2.0.0-bin-hadoop2.6/python/pyspark/sql/utils.pyc in deco(*a, **kw) 
    61  def deco(*a, **kw): 
    62   try: 
---> 63    return f(*a, **kw) 
    64   except py4j.protocol.Py4JJavaError as e: 
    65    s = e.java_exception.toString() 

/usr/local/lib/python2.7/dist-packages/py4j/protocol.pyc in get_return_value(answer, gateway_client, target_id, name) 
    317     raise Py4JJavaError(
    318      "An error occurred while calling {0}{1}{2}.\n". 
--> 319      format(target_id, ".", name), value) 
    320    else: 
    321     raise Py4JError(

Py4JJavaError: An error occurred while calling o321.load. 
: org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: file:[filename]/metadata 

`` `

Gibt es eine einfache Behelfslösung für lade das? Das Format des LIBSVM-Modells ist

x 
0 1 
1 -1050 1030 
2 0 1 
3 0 3 
4 0 1 
5 0 1 

Answer 1

Erstens ist die Datei nicht im libsvm-Format dargestellt. Das korrekte Format einer libsvm-Datei ist das folgende:

<label> <index1>:<value1> <index2>:<value2> ... <indexN>:<valueN> 

Daher ist Ihre Datenvorbereitung zu Beginn falsch.

Zweitens liest die Klassenmethode load(path), die Sie mit MinMaxScaler verwenden, eine ML-Instanz aus dem Eingabepfad.

Denken Sie daran, dass:MinMaxScaler Auswertungsstatistiken auf einem Datensatz berechnet und erzeugt ein MinMaxScalerModel. Das Modell kann dann jedes Merkmal einzeln so transformieren, dass es in dem gegebenen Bereich liegt.

z:

from pyspark.ml.linalg import Vectors 
from pyspark.mllib.regression import LabeledPoint 
from pyspark.ml.feature import MinMaxScaler 
df = spark.createDataFrame([(1.1, Vectors.sparse(3, [(0, 1.23), (2, 4.56)])) ,(0.0, Vectors.dense([1.01, 2.02, 3.03]))],['label','features']) 

df.show(truncate=False) 
# +-----+---------------------+ 
# |label|features    | 
# +-----+---------------------+ 
# |1.1 |(3,[0,2],[1.23,4.56])| 
# |0.0 |[1.01,2.02,3.03]  | 
# +-----+---------------------+ 

mmScaler = MinMaxScaler(inputCol="features", outputCol="scaled") 
temp_path = "/tmp/spark/" 
minMaxScalerPath = temp_path + "min-max-scaler" 
mmScaler.save(minMaxScalerPath) 

Das Snippet oben wird die MinMaxScaler Funktion Transformator speichern, so kann es nach der Klassenmethode Last geladen werden.

Schauen wir uns an, was tatsächlich passiert ist. Die Klassenmethode save die folgende Dateistruktur erstellen:

/tmp/spark/ 
└── min-max-scaler 
    └── metadata 
     ├── part-00000 
     └── _SUCCESS 

Lassen Sie uns den Inhalt dieser part-0000 Datei:

$ cat /tmp/spark/min-max-scaler/metadata/part-00000 | python -m json.tool 
{ 
    "class": "org.apache.spark.ml.feature.MinMaxScaler", 
    "paramMap": { 
     "inputCol": "features", 
     "max": 1.0, 
     "min": 0.0, 
     "outputCol": "scaled" 
    }, 
    "sparkVersion": "2.0.0", 
    "timestamp": 1480501003244, 
    "uid": "MinMaxScaler_42e68455a929c67ba66f" 
} 

Also eigentlich, wenn man den Transformator laden:

loadedMMScaler = MinMaxScaler.load(minMaxScalerPath) 

Sie laden diese Datei tatsächlich. Es wird keine libsvm Datei benötigt!

Jetzt können Sie Ihren Transformator anwenden, das Modell zu erstellen und verwandeln Sie Ihre DataFrame:

model = loadedMMScaler.fit(df) 

model.transform(df).show(truncate=False)          
# +-----+---------------------+-------------+ 
# |label|features    |scaled  | 
# +-----+---------------------+-------------+ 
# |1.1 |(3,[0,2],[1.23,4.56])|[1.0,0.0,1.0]| 
# |0.0 |[1.01,2.02,3.03]  |[0.0,1.0,0.0]| 
# +-----+---------------------+-------------+ 

Nun zurück zu dieser LIBSVM Datei erhalten lassen und lassen Sie uns einige Dummy-Daten erstellen und in ein LIBSVM Format speichern mit MLUtils

from pyspark.mllib.regression import LabeledPoint 
from pyspark.mllib.linalg import Vectors 
from pyspark.mllib.util import MLUtils 
data = sc.parallelize([LabeledPoint(1.1, Vectors.sparse(3, [(0, 1.23), (2, 4.56)])), LabeledPoint(0.0, Vectors.dense([1.01, 2.02, 3.03]))]) 
MLUtils.saveAsLibSVMFile(data, temp_path + "data") 

Zurück zu unserer Dateistruktur:

/tmp/spark/ 
├── data 
│   ├── part-00000 
│   ├── part-00001 
│   ├── part-00002 
│   ├── part-00003 
│   ├── part-00004 
│   ├── part-00005 
│   ├── part-00006 
│   ├── part-00007 
│   └── _SUCCESS 
└── min-max-scaler 
    └── metadata 
     ├── part-00000 
     └── _SUCCESS 

können Sie überprüfen den Inhalt der Datei, die jetzt in LIBSVM Format ist:

$ cat /tmp/spark/data/part-0000* 
1.1 1:1.23 3:4.56 
0.0 1:1.01 2:2.02 3:3.03 

Nun wollen wir, dass die Daten laden und anwenden:

loadedData = MLUtils.loadLibSVMFile(sc, temp_path + "data") 
loadedDataDF = spark.createDataFrame(loadedData.map(lambda lp : (lp.label, lp.features.asML())), ['label','features']) 

loadedDataDF.show(truncate=False) 
# +-----+----------------------------+            
# |label|features     | 
# +-----+----------------------------+ 
# |1.1 |(3,[0,2],[1.23,4.56])  | 
# |0.0 |(3,[0,1,2],[1.01,2.02,3.03])| 
# +-----+----------------------------+ 

Hinweis dass MLlib Umwandlung VectorsVectors bis ML ist sehr wichtig. Sie können mehr darüber lesen here.

model.transform(loadedDataDF).show(truncate=False) 
# +-----+----------------------------+-------------+ 
# |label|features     |scaled  | 
# +-----+----------------------------+-------------+ 
# |1.1 |(3,[0,2],[1.23,4.56])  |[1.0,0.0,1.0]| 
# |0.0 |(3,[0,1,2],[1.01,2.02,3.03])|[0.0,1.0,0.0]| 
# +-----+----------------------------+-------------+ 

Ich hoffe, dass dies Ihre Frage beantwortet!