XGBoost-4j von DMLC auf Spark-1.6.1

Question

Ich versuche, die XGBoost-Implementierung von DMLC auf Spark-1.6.1 zu verwenden. Ich kann meine Daten mit XGBoost trainieren, habe aber Schwierigkeiten bei der Vorhersage. Ich möchte tatsächlich eine Vorhersage machen, wie es in Apache Spark mllib Bibliotheken gemacht werden kann, die bei der Berechnung von Trainingsfehlern, Präzision, Rückruf, Spezifität usw. hilft.

Ich poste die Codes unten, auch der Fehler bin ich bekommen. Ich habe diese xgboost4j-spark-0.5-jar-with-dependencies.jar in Spark-Shell zu starten.

import org.apache.spark.mllib.regression.LabeledPoint 
import org.apache.spark.mllib.linalg.Vectors 
import org.apache.spark.mllib.util.MLUtils 
import org.apache.spark.SparkContext._ 
import org.apache.spark.mllib.linalg.Vectors 
import org.apache.spark.mllib.regression.LabeledPoint 
import ml.dmlc.xgboost4j.scala.Booster 
import ml.dmlc.xgboost4j.scala.spark.XGBoost 
import ml.dmlc.xgboost4j.scala.DMatrix 
import ml.dmlc.xgboost4j.scala.{Booster, DMatrix} 
import ml.dmlc.xgboost4j.scala.spark.{DataUtils, XGBoost} 
import org.apache.spark.{SparkConf, SparkContext} 




//Load and parse the data file. 
val data = sc.textFile("file:///home/partha/credit_approval_2_attr.csv") 
val data1 = sc.textFile("file:///home/partha/credit_app_fea.csv") 


val parsedData = data.map { line => 
    val parts = line.split(',').map(_.toDouble) 
    LabeledPoint(parts(0), Vectors.dense(parts.tail)) 
}.cache() 

val parsedData1 = data1.map { line => 
    val parts = line.split(',').map(_.toDouble) 
    Vectors.dense(parts) 
} 



//Tuning Parameters 
val paramMap = List(
     "eta" -> 0.1f, 
     "max_depth" -> 5, 
     "num_class" -> 2, 
     "objective" -> "multi:softmax" , 
     "colsample_bytree" -> 0.8, 
     "alpha" -> 1, 
     "subsample" -> 0.5).toMap 

    //Training the model 
val numRound = 20 
val model = XGBoost.train(parsedData, paramMap, numRound, nWorkers = 1) 
val pred = model.predict(parsedData1) 
pred.collect() 

Ausgabe von pred:

res0: Array[Array[Array[Float]]] = Array(Array(Array(0.0), Array(1.0), Array(1.0), Array(1.0), Array(0.0), Array(0.0), Array(1.0), Array(1.0), Array(0.0), Array(1.0), Array(0.0), Array(0.0), Array(0.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(0.0), Array(1.0), Array(1.0), Array(0.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(0.0), Array(1.0), Array(1.0), Array(1.0), Array(0.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(1.0), Array(0.0), Array(0.0), Array(0.0), Array(0.0), Array(1.0), Array(0.0), Array(0.0), Array(0.0), Array(0.0), Array(0.0), Array(0.0), Array(1.0), Array(1.0), Array(1.0), Array(... 

Nun, wenn ich verwende:

val labelAndPreds = parsedData.map { point => 
      val prediction = model.predict(point.features) 
      (point.label, prediction) 
     } 

Ausgang:

<console>:66: error: overloaded method value predict with alternatives: 
    (testSet: ml.dmlc.xgboost4j.scala.DMatrix)Array[Array[Float]] <and> 
    (testSet: org.apache.spark.rdd.RDD[org.apache.spark.mllib.linalg.Vector])org.apache.spark.rdd.RDD[Array[Array[Float]]] 
cannot be applied to (org.apache.spark.mllib.linalg.Vector) 
        val prediction = model.predict(point.features) 
            ^

Und dann versucht, diese, da vorhersagen erfordert eine RDD [Vektor].

val labelAndPreds1 = parsedData.map { point => 
      val prediction = model.predict(Vectors.dense(point.features)) 
      (point.label, prediction) 
     } 

Das Ergebnis war:

<console>:66: error: overloaded method value dense with alternatives: 
    (values: Array[Double])org.apache.spark.mllib.linalg.Vector <and> 
    (firstValue: Double,otherValues: Double*)org.apache.spark.mllib.linalg.Vector 
cannot be applied to (org.apache.spark.mllib.linalg.Vector) 
        val prediction = model.predict(Vectors.dense(point.features)) 
                 ^

klar sein eine Frage der RDD-Typ, den ich zu sortieren, ich versuche, dies mit GBT auf Funken leicht (http://spark.apache.org/docs/latest/mllib-ensembles.html#gradient-boosted-trees-gbts).

Mache ich das auf die richtige Weise?

Jede Hilfe oder Anregung wäre genial.

Answer 1

, wenn Sie lesen, den Quellcode vorhersagen()

#

def predict(testSet: RDD[Vector]): RDD[Array[Array[Float]]] = { 
    import DataUtils._ 
    val broadcastBooster = testSet.sparkContext.broadcast(_booster) 
    testSet.mapPartitions { testSamples => 
     if (testSamples.hasNext) { 
     val dMatrix = new DMatrix(new JDMatrix(testSamples, null)) 
     Iterator(broadcastBooster.value.predict(dMatrix)) 
     } else { 
     Iterator() 
     } 
    } 
    } 

#

Sie testSet.mapPartitions() auf dem Testdaten finden würde, ist das Ergebnis Array von Array, die innere Anordnung ist der Vorhersagewert für Testdaten. Sie sollten FlatMap auf das Ergebnis tun.

Answer 2

Tatsächlich ist dies bei XGboost-Algorithmen nicht verfügbar. Ich bin vor dem gleichen Problem hier und haben die folgende Methode implementiert:

import ml.dmlc.xgboost4j.scala.spark.DataUtils // thanks to @Z Simon 

def labelPredict(testSet: RDD[XGBLabeledPoint], 
       useExternalCache: Boolean = false, 
       booster: XGBoostModel): RDD[(Float, Float)] = { 
val broadcastBooster = testSet.sparkContext.broadcast(booster) 
testSet.mapPartitions { testData => 
    val (auxiliaryIterator, testDataIterator) = testData.duplicate 
    val testDataArray = auxiliaryIterator.toArray 
    val prediction = broadcastBooster.value.predict(new DMatrix(testDataIterator)).flatten 
    testDataArray 
    .zip(prediction) 
    .map { 
     case (labeledPoint, predictionValue) => 
     (labeledPoint.label, predictionValue) 
    }.toIterator 
} 

}

Dies ist fast der gleiche Code, der XGBoost tatsächlich hat, aber es ist das Label von labeledpoint in den Vorhersagen unter Verwendung von Rückkehr. Wenn Sie einen Labeledpoint an diese Methode übergeben, wird eine RDD von Tuple mit (Label, Prädiktion) für jeden Wert zurückgegeben.