Hendrik/xgboost
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

allow empty partitions

Browse Source
This commit is contained in:
CodingCat 2016-03-23 12:30:06 -04:00
parent f1114688a7
commit d8535313eb
2 changed files with 27 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
jvm-packages/xgboost4j-spark/src/main/scala/ml/dmlc/xgboost4j/scala/spark/XGBoostModel.scala
View File

@ -32,8 +32,12 @@ class XGBoostModel(_booster: Booster)(implicit val sc: SparkContext) extends Ser
import DataUtils._ import DataUtils._
val broadcastBooster = testSet.sparkContext.broadcast(_booster) val broadcastBooster = testSet.sparkContext.broadcast(_booster)
testSet.mapPartitions { testSamples => testSet.mapPartitions { testSamples =>
val dMatrix = new DMatrix(new JDMatrix(testSamples, null)) if (testSamples.hasNext) {
Iterator(broadcastBooster.value.predict(dMatrix)) val dMatrix = new DMatrix(new JDMatrix(testSamples, null))
Iterator(broadcastBooster.value.predict(dMatrix))
} else {
Iterator()
}
} }
} }

22
jvm-packages/xgboost4j-spark/src/test/scala/ml/dmlc/xgboost4j/scala/spark/XGBoostSuite.scala
View File

@ -23,7 +23,7 @@ import scala.collection.mutable.ListBuffer
import scala.io.Source import scala.io.Source
import org.apache.commons.logging.LogFactory import org.apache.commons.logging.LogFactory
import org.apache.spark.mllib.linalg.DenseVector import org.apache.spark.mllib.linalg.{Vector => SparkVector, Vectors, DenseVector}
import org.apache.spark.mllib.regression.LabeledPoint import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.rdd.RDD import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.{SparkConf, SparkContext}
@ -190,4 +190,24 @@ class XGBoostSuite extends FunSuite with BeforeAndAfter {
assert(eval.eval(xgBoostModel.predict(testSetDMatrix), testSetDMatrix) < 0.1) assert(eval.eval(xgBoostModel.predict(testSetDMatrix), testSetDMatrix) < 0.1)
customSparkContext.stop() customSparkContext.stop()
} }
test("test with empty partition") {
def buildEmptyRDD(sparkContext: Option[SparkContext] = None): RDD[SparkVector] = {
val sampleList = new ListBuffer[SparkVector]
sparkContext.getOrElse(sc).parallelize(sampleList, numWorkers)
}
val eval = new EvalError()
val trainingRDD = buildTrainingRDD()
val testRDD = buildEmptyRDD()
import DataUtils._
val tempDir = Files.createTempDirectory("xgboosttest-")
val tempFile = Files.createTempFile(tempDir, "", "")
val paramMap = List("eta" -> "1", "max_depth" -> "2", "silent" -> "0",
"objective" -> "binary:logistic").toMap
val xgBoostModel = XGBoost.train(trainingRDD, paramMap, 5, numWorkers)
println(xgBoostModel.predict(testRDD))
}
} }