diff --git a/mllib/src/main/scala/org/apache/spark/mllib/tree/DecisionTree.scala b/mllib/src/main/scala/org/apache/spark/mllib/tree/DecisionTree.scala index 73e7e32c6db31..82182e43b08fe 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/tree/DecisionTree.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/tree/DecisionTree.scala @@ -17,6 +17,10 @@ package org.apache.spark.mllib.tree +import java.util.Random + +import org.apache.spark.util.Utils + import scala.collection.JavaConverters._ import scala.collection.mutable import scala.collection.mutable.ArrayBuffer @@ -60,11 +64,20 @@ class DecisionTree (private val strategy: Strategy) extends Serializable with Lo */ def run(input: RDD[LabeledPoint]): DecisionTreeModel = { // Note: random seed will not be used since numTrees = 1. - val rf = new RandomForest(strategy, numTrees = 1, featureSubsetStrategy = "all", seed = 0) + val rf = new RandomForest(strategy, numTrees = 1, featureSubsetStrategy = x => x, seed = 0) + val rfModel = rf.run(input) + rfModel.trees(0) + } + + + def run(input: RDD[LabeledPoint], featureSubsetStrategy: Int => Int ): DecisionTreeModel = { + val rf = new RandomForest(strategy, numTrees = 1, featureSubsetStrategy, + seed = Utils.random.nextInt) val rfModel = rf.run(input) rfModel.trees(0) } + /** * Trains a decision tree model over an RDD. This is deprecated because it hides the static * methods with the same name in Java. diff --git a/mllib/src/main/scala/org/apache/spark/mllib/tree/GradientBoostedTrees.scala b/mllib/src/main/scala/org/apache/spark/mllib/tree/GradientBoostedTrees.scala index 61f6b1313f82e..44312e9b973bc 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/tree/GradientBoostedTrees.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/tree/GradientBoostedTrees.scala @@ -25,7 +25,7 @@ import org.apache.spark.mllib.tree.configuration.BoostingStrategy import org.apache.spark.mllib.tree.configuration.Algo._ import org.apache.spark.mllib.tree.impl.TimeTracker import org.apache.spark.mllib.tree.impurity.Variance -import org.apache.spark.mllib.tree.model.{DecisionTreeModel, GradientBoostedTreesModel} +import org.apache.spark.mllib.tree.model.{Predict, Node, DecisionTreeModel, GradientBoostedTreesModel} import org.apache.spark.rdd.RDD import org.apache.spark.storage.StorageLevel @@ -113,7 +113,8 @@ object GradientBoostedTrees extends Logging { */ private def boost( input: RDD[LabeledPoint], - boostingStrategy: BoostingStrategy): GradientBoostedTreesModel = { + boostingStrategy: BoostingStrategy, + featureSubsetSelection: Int => Int = x => x): GradientBoostedTreesModel = { val timer = new TimeTracker() timer.start("total") @@ -147,16 +148,29 @@ object GradientBoostedTrees extends Logging { // Initialize tree timer.start("building tree 0") - val firstTreeModel = new DecisionTree(treeStrategy).run(data) + + + val avgPos = input.map(_.label).collect.sum/input.count + val startingHalfLogOdds = 0.5 * Math.log((1 + avgPos)/(1 - avgPos)) + val topNode = Node.apply(0,new Predict(startingHalfLogOdds,startingHalfLogOdds),0,true) + val firstTreeModel = new DecisionTreeModel(topNode, Regression) baseLearners(0) = firstTreeModel baseLearnerWeights(0) = 1.0 - val startingModel = new GradientBoostedTreesModel(Regression, Array(firstTreeModel), Array(1.0)) - logDebug("error of gbt = " + loss.computeError(startingModel, input)) + + + //val firstTreeModel = new DecisionTree(treeStrategy).run(data, + // boostingStrategy.featureSubsetSelection) + //baseLearners(0) = firstTreeModel + //baseLearnerWeights(0) = 1.0 + + var partialModel = + new GradientBoostedTreesModel(Regression, Array(firstTreeModel), Array(1.0)) + logDebug("error of gbt = " + loss.computeError(partialModel, input)) // Note: A model of type regression is used since we require raw prediction timer.stop("building tree 0") - // psuedo-residual for second iteration - data = input.map(point => LabeledPoint(loss.gradient(startingModel, point), + // pseudo-residual for second iteration + data = input.map(point => LabeledPoint(-loss.gradient(partialModel, point), point.features)) var m = 1 @@ -165,7 +179,9 @@ object GradientBoostedTrees extends Logging { logDebug("###################################################") logDebug("Gradient boosting tree iteration " + m) logDebug("###################################################") - val model = new DecisionTree(treeStrategy).run(data) + val model = new DecisionTree(treeStrategy).run(data, + boostingStrategy.featureSubsetSelection) + updateTree(model,partialModel,input) timer.stop(s"building tree $m") // Create partial model baseLearners(m) = model @@ -174,7 +190,7 @@ object GradientBoostedTrees extends Logging { // However, the behavior should be reasonable, though not optimal. baseLearnerWeights(m) = learningRate // Note: A model of type regression is used since we require raw prediction - val partialModel = new GradientBoostedTreesModel( + partialModel = new GradientBoostedTreesModel( Regression, baseLearners.slice(0, m + 1), baseLearnerWeights.slice(0, m + 1)) logDebug("error of gbt = " + loss.computeError(partialModel, input)) // Update data with pseudo-residuals @@ -191,4 +207,38 @@ object GradientBoostedTrees extends Logging { new GradientBoostedTreesModel( boostingStrategy.treeStrategy.algo, baseLearners, baseLearnerWeights) } + + def updateTree(tree: DecisionTreeModel, partialModel: GradientBoostedTreesModel, + input: RDD[LabeledPoint]) { + val triples = input.map(point => (point.label,tree.getNodeId(point.features), + partialModel.predict(point.features))) + .groupBy(triple => triple._2).collect + val newData = triples.map(newValue) + for (d <- newData) { + val id = d._1 + val newVal = d._2 + Node.getNode(id,tree.topNode).predict = new Predict(newVal,newVal) + } + } + + def newValue(nodeData: (Int,Iterable[(Double,Int,Double)])): (Int,Double) = { + val id = nodeData._1 + val data = nodeData._2 + val pseudoResponses = data.toArray.map{case (label,id,prediction) => + 2.0 * label/(1.0 + Math.exp(2.0 * label * prediction))} + + + val numerator = pseudoResponses.sum + var denominator = 0.0 + for (pr <- pseudoResponses) { + denominator = denominator + Math.abs(pr) * (2.0 - Math.abs(pr)) + } + + //(id, pseudoResponses.sum/pseudoResponses.size) + (id, numerator/denominator) + } + + + + } diff --git a/mllib/src/main/scala/org/apache/spark/mllib/tree/RandomForest.scala b/mllib/src/main/scala/org/apache/spark/mllib/tree/RandomForest.scala index e9304b5e5c650..73b3f4e3e0c5d 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/tree/RandomForest.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/tree/RandomForest.scala @@ -67,7 +67,7 @@ import org.apache.spark.util.Utils private class RandomForest ( private val strategy: Strategy, private val numTrees: Int, - featureSubsetStrategy: String, + featureSubsetStrategy: Int => Int, private val seed: Int) extends Serializable with Logging { @@ -114,9 +114,12 @@ private class RandomForest ( strategy.assertValid() require(numTrees > 0, s"RandomForest requires numTrees > 0, but was given numTrees = $numTrees.") + + /* require(RandomForest.supportedFeatureSubsetStrategies.contains(featureSubsetStrategy), s"RandomForest given invalid featureSubsetStrategy: $featureSubsetStrategy." + s" Supported values: ${RandomForest.supportedFeatureSubsetStrategies.mkString(", ")}.") + */ /** * Method to train a decision tree model over an RDD @@ -155,6 +158,8 @@ private class RandomForest ( // Cache input RDD for speedup during multiple passes. val treeInput = TreePoint.convertToTreeRDD(retaggedInput, bins, metadata) + + // FIX ME SO THAT WE CAN SUBSAMPLE SINGLE TREES FOR GBDT val (subsample, withReplacement) = { // TODO: Have a stricter check for RF in the strategy val isRandomForest = numTrees > 1 @@ -284,7 +289,8 @@ object RandomForest extends Serializable with Logging { seed: Int): RandomForestModel = { require(strategy.algo == Classification, s"RandomForest.trainClassifier given Strategy with invalid algo: ${strategy.algo}") - val rf = new RandomForest(strategy, numTrees, featureSubsetStrategy, seed) + val rf = new RandomForest(strategy, numTrees, DecisionTreeMetadata.buildFeatureSubsetStrategy( + featureSubsetStrategy, numTrees, strategy.algo), seed) rf.run(input) } @@ -366,7 +372,7 @@ object RandomForest extends Serializable with Logging { input: RDD[LabeledPoint], strategy: Strategy, numTrees: Int, - featureSubsetStrategy: String, + featureSubsetStrategy: Int => Int, seed: Int): RandomForestModel = { require(strategy.algo == Regression, s"RandomForest.trainRegressor given Strategy with invalid algo: ${strategy.algo}") @@ -402,11 +408,11 @@ object RandomForest extends Serializable with Logging { input: RDD[LabeledPoint], categoricalFeaturesInfo: Map[Int, Int], numTrees: Int, - featureSubsetStrategy: String, + featureSubsetStrategy: Int => Int, impurity: String, maxDepth: Int, maxBins: Int, - seed: Int = Utils.random.nextInt()): RandomForestModel = { + seed: Int): RandomForestModel = { val impurityType = Impurities.fromString(impurity) val strategy = new Strategy(Regression, impurityType, maxDepth, 0, maxBins, Sort, categoricalFeaturesInfo) @@ -430,6 +436,76 @@ object RandomForest extends Serializable with Logging { numTrees, featureSubsetStrategy, impurity, maxDepth, maxBins, seed) } + + /** + * Method to train a decision tree model for regression. + * + * @param input Training dataset: RDD of [[org.apache.spark.mllib.regression.LabeledPoint]]. + * Labels are real numbers. + * @param strategy Parameters for training each tree in the forest. + * @param numTrees Number of trees in the random forest. + * @param featureSubsetStrategy Number of features to consider for splits at each node. + * Supported: "auto", "all", "sqrt", "log2", "onethird". + * If "auto" is set, this parameter is set based on numTrees: + * if numTrees == 1, set to "all"; + * if numTrees > 1 (forest) set to "onethird". + * @param seed Random seed for bootstrapping and choosing feature subsets. + * @return a random forest model that can be used for prediction + */ + def trainRegressor( + input: RDD[LabeledPoint], + strategy: Strategy, + numTrees: Int, + featureSubsetStrategy: String, + seed: Int): RandomForestModel = { + require(strategy.algo == Regression, + s"RandomForest.trainRegressor given Strategy with invalid algo: ${strategy.algo}") + val rf = new RandomForest(strategy, numTrees, + DecisionTreeMetadata.buildFeatureSubsetStrategy(featureSubsetStrategy, + numTrees, strategy.algo), seed) + rf.run(input) + } + + /** + * Method to train a decision tree model for regression. + * + * @param input Training dataset: RDD of [[org.apache.spark.mllib.regression.LabeledPoint]]. + * Labels are real numbers. + * @param categoricalFeaturesInfo Map storing arity of categorical features. + * E.g., an entry (n -> k) indicates that feature n is categorical + * with k categories indexed from 0: {0, 1, ..., k-1}. + * @param numTrees Number of trees in the random forest. + * @param featureSubsetStrategy Number of features to consider for splits at each node. + * Supported: "auto", "all", "sqrt", "log2", "onethird". + * If "auto" is set, this parameter is set based on numTrees: + * if numTrees == 1, set to "all"; + * if numTrees > 1 (forest) set to "onethird". + * @param impurity Criterion used for information gain calculation. + * Supported values: "variance". + * @param maxDepth Maximum depth of the tree. + * E.g., depth 0 means 1 leaf node; depth 1 means 1 internal node + 2 leaf nodes. + * (suggested value: 4) + * @param maxBins maximum number of bins used for splitting features + * (suggested value: 100) + * @param seed Random seed for bootstrapping and choosing feature subsets. + * @return a random forest model that can be used for prediction + */ + def trainRegressor( + input: RDD[LabeledPoint], + categoricalFeaturesInfo: Map[Int, Int], + numTrees: Int, + featureSubsetStrategy: String, + impurity: String, + maxDepth: Int, + maxBins: Int, + seed: Int = Utils.random.nextInt()): RandomForestModel = { + val impurityType = Impurities.fromString(impurity) + val strategy = new Strategy(Regression, impurityType, maxDepth, + 0, maxBins, Sort, categoricalFeaturesInfo) + trainRegressor(input, strategy, numTrees, featureSubsetStrategy, seed) + } + + /** * List of supported feature subset sampling strategies. */ diff --git a/mllib/src/main/scala/org/apache/spark/mllib/tree/configuration/BoostingStrategy.scala b/mllib/src/main/scala/org/apache/spark/mllib/tree/configuration/BoostingStrategy.scala index cf51d041c65a9..e0159513e3b3f 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/tree/configuration/BoostingStrategy.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/tree/configuration/BoostingStrategy.scala @@ -42,8 +42,8 @@ case class BoostingStrategy( @BeanProperty var loss: Loss, // Optional boosting parameters @BeanProperty var numIterations: Int = 100, - @BeanProperty var learningRate: Double = 0.1) extends Serializable { - + @BeanProperty var learningRate: Double = 0.1, + @BeanProperty var featureSubsetSelection: Int => Int = x => x) extends Serializable { /** * Check validity of parameters. * Throws exception if invalid. diff --git a/mllib/src/main/scala/org/apache/spark/mllib/tree/impl/DecisionTreeMetadata.scala b/mllib/src/main/scala/org/apache/spark/mllib/tree/impl/DecisionTreeMetadata.scala index 951733fada6be..8fd4494f3e9b9 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/tree/impl/DecisionTreeMetadata.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/tree/impl/DecisionTreeMetadata.scala @@ -96,6 +96,44 @@ private[tree] class DecisionTreeMetadata( private[tree] object DecisionTreeMetadata extends Logging { + def buildFeatureSubsetStrategy(featureSubsetStrategy: String, + numTrees: Int, algo: Algo): Int => Int = { + + val _featureSubsetStrategy = featureSubsetStrategy match { + case "auto" => + if (numTrees == 1) { + "all" + } else { + if (algo == Classification) { + "sqrt" + } else { + "onethird" + } + } + case _ => featureSubsetStrategy + } + + val numFeaturesPerNodeRule: Int => Int = _featureSubsetStrategy match { + case "all" => x => x + case "sqrt" => x => math.sqrt(x).ceil.toInt + case "log2" => x => math.max(1, (math.log(x) / math.log(2)).ceil.toInt) + case "onethird" => x => (x / 3.0).ceil.toInt + } + numFeaturesPerNodeRule + } + + def buildMetadata ( + input: RDD[LabeledPoint], + strategy: Strategy, + numTrees: Int, + featureSubsetStrategy: String): DecisionTreeMetadata = { + + val numFeaturesPerNodeRule = buildFeatureSubsetStrategy(featureSubsetStrategy, + numTrees, strategy.algo) + buildMetadata(input, strategy, numTrees, numFeaturesPerNodeRule) + } + + /** * Construct a [[DecisionTreeMetadata]] instance for this dataset and parameters. * This computes which categorical features will be ordered vs. unordered, @@ -105,7 +143,7 @@ private[tree] object DecisionTreeMetadata extends Logging { input: RDD[LabeledPoint], strategy: Strategy, numTrees: Int, - featureSubsetStrategy: String): DecisionTreeMetadata = { + featureSubsetStrategy: Int => Int): DecisionTreeMetadata = { val numFeatures = input.take(1)(0).features.size val numExamples = input.count() @@ -155,26 +193,7 @@ private[tree] object DecisionTreeMetadata extends Logging { } } - // Set number of features to use per node (for random forests). - val _featureSubsetStrategy = featureSubsetStrategy match { - case "auto" => - if (numTrees == 1) { - "all" - } else { - if (strategy.algo == Classification) { - "sqrt" - } else { - "onethird" - } - } - case _ => featureSubsetStrategy - } - val numFeaturesPerNode: Int = _featureSubsetStrategy match { - case "all" => numFeatures - case "sqrt" => math.sqrt(numFeatures).ceil.toInt - case "log2" => math.max(1, (math.log(numFeatures) / math.log(2)).ceil.toInt) - case "onethird" => (numFeatures / 3.0).ceil.toInt - } + val numFeaturesPerNode = featureSubsetStrategy(numFeatures) new DecisionTreeMetadata(numFeatures, numExamples, numClasses, numBins.max, strategy.categoricalFeaturesInfo, unorderedFeatures.toSet, numBins, diff --git a/mllib/src/main/scala/org/apache/spark/mllib/tree/loss/LogLoss.scala b/mllib/src/main/scala/org/apache/spark/mllib/tree/loss/LogLoss.scala index 98d8a2f251aeb..be91d33c21187 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/tree/loss/LogLoss.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/tree/loss/LogLoss.scala @@ -47,7 +47,7 @@ object LogLoss extends Loss { model: TreeEnsembleModel, point: LabeledPoint): Double = { val prediction = model.predict(point.features) - - 4.0 * point.label / (1.0 + math.exp(2.0 * point.label * prediction)) + -2.0 * point.label / (1.0 + math.exp(2.0 * point.label * prediction)) } /** @@ -71,4 +71,5 @@ object LogLoss extends Loss { } }.mean() } + } diff --git a/mllib/src/main/scala/org/apache/spark/mllib/tree/model/DecisionTreeModel.scala b/mllib/src/main/scala/org/apache/spark/mllib/tree/model/DecisionTreeModel.scala index a5760963068c3..28cff27f95df7 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/tree/model/DecisionTreeModel.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/tree/model/DecisionTreeModel.scala @@ -43,6 +43,10 @@ class DecisionTreeModel(val topNode: Node, val algo: Algo) extends Serializable topNode.predict(features) } + def getNodeId(features: Vector): Int= { + topNode.getNodeId(features) + } + /** * Predict values for the given data set using the model trained. * @@ -99,4 +103,6 @@ class DecisionTreeModel(val topNode: Node, val algo: Algo) extends Serializable header + topNode.subtreeToString(2) } + + } diff --git a/mllib/src/main/scala/org/apache/spark/mllib/tree/model/Node.scala b/mllib/src/main/scala/org/apache/spark/mllib/tree/model/Node.scala index 2179da8dbe03e..ab2f136296627 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/tree/model/Node.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/tree/model/Node.scala @@ -98,6 +98,26 @@ class Node ( } } + def getNodeId(features: Vector) : Int = { + if (isLeaf) { + id + } else{ + if (split.get.featureType == Continuous) { + if (features(split.get.feature) <= split.get.threshold) { + leftNode.get.getNodeId(features) + } else { + rightNode.get.getNodeId(features) + } + } else { + if (split.get.categories.contains(features(split.get.feature))) { + leftNode.get.getNodeId(features) + } else { + rightNode.get.getNodeId(features) + } + } + } + } + /** * Returns a deep copy of the subtree rooted at this node. */