minor work done for click prediction

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.

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)
+  }
+
+
+
+
 }

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.
    */

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.

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,

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()
   }
+
 }