import org.apache.spark.mllib.tree.DecisionTreeimport org.apache.spark.mllib.util.MLUtils// Load and parse the data file.// Cache the data since we will use it again to compute training error.val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt").cache()// Train a DecisionTree model.// Empty categoricalFeaturesInfo indicates all features are continuous.val categoricalFeaturesInfo = Map[Int, Int]()val impurity = "variance"val maxDepth = 5val maxBins = 100val model = DecisionTree.trainRegressor(data, categoricalFeaturesInfo, impurity,  maxDepth, maxBins)// Evaluate model on training instances and compute training errorval labelsAndPredictions = data.map { point =>  val prediction = model.predict(point.features)  (point.label, prediction)}val trainMSE = labelsAndPredictions.map{ case(v, p) => math.pow((v - p), 2)}.mean()println("Training Mean Squared Error = " + trainMSE)println("Learned regression tree model:\n" + model)

/**   * 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 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: 5)   * @param maxBins maximum number of bins used for splitting features   *                 (suggested value: 32)   * @return DecisionTreeModel that can be used for prediction   */  def trainRegressor(      input: RDD[LabeledPoint],      categoricalFeaturesInfo: Map[Int, Int],      impurity: String,      maxDepth: Int,      maxBins: Int): DecisionTreeModel = {    val impurityType = Impurities.fromString(impurity)    train(input, Regression, impurityType, maxDepth, 0, maxBins, Sort, categoricalFeaturesInfo)  }

def train(      input: RDD[LabeledPoint],      algo: Algo,      impurity: Impurity,      maxDepth: Int,      numClassesForClassification: Int,      maxBins: Int,      quantileCalculationStrategy: QuantileStrategy,      categoricalFeaturesInfo: Map[Int,Int]): DecisionTreeModel = {    val strategy = new Strategy(algo, impurity, maxDepth, numClassesForClassification, maxBins,      quantileCalculationStrategy, categoricalFeaturesInfo)    new DecisionTree(strategy).train(input)  }

/** * :: Experimental :: * A class which implements a decision tree learning algorithm for classification and regression. * It supports both continuous and categorical features. * @param strategy The configuration parameters for the tree algorithm which specify the type *                 of algorithm (classification, regression, etc.), feature type (continuous, *                 categorical), depth of the tree, quantile calculation strategy, etc. */@Experimentalclass DecisionTree (private val strategy: Strategy) extends Serializable with Logging {  strategy.assertValid()  /**   * Method to train a decision tree model over an RDD   * @param input Training data: RDD of [[org.apache.spark.mllib.regression.LabeledPoint]]   * @return DecisionTreeModel that can be used for prediction   */  def train(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 rfModel = rf.train(input)    rfModel.trees(0)  }}

/**   * Method to train a decision tree model over an RDD   * @param input Training data: RDD of [[org.apache.spark.mllib.regression.LabeledPoint]]   * @return RandomForestModel that can be used for prediction   */  def train(input: RDD[LabeledPoint]): RandomForestModel = {   //1. metadata    val retaggedInput = input.retag(classOf[LabeledPoint])    val metadata =      DecisionTreeMetadata.buildMetadata(retaggedInput, strategy, numTrees, featureSubsetStrategy)    // 2. Find the splits and the corresponding bins (interval between the splits) using a sample    // of the input data.    val (splits, bins) = DecisionTree.findSplitsBins(retaggedInput, metadata)    // 3. Bin feature values (TreePoint representation).    // Cache input RDD for speedup during multiple passes.    val treeInput = TreePoint.convertToTreeRDD(retaggedInput, bins, metadata)    val baggedInput = if (numTrees > 1) {      BaggedPoint.convertToBaggedRDD(treeInput, numTrees, seed)    } else {      BaggedPoint.convertToBaggedRDDWithoutSampling(treeInput)    }.persist(StorageLevel.MEMORY_AND_DISK)    // set maxDepth and compute memory usage     // depth of the decision tree    // Max memory usage for aggregates    // TODO: Calculate memory usage more precisely.    //........    /*     * The main idea here is to perform group-wise training of the decision tree nodes thus     * reducing the passes over the data from (# nodes) to (# nodes / maxNumberOfNodesPerGroup).     * Each data sample is handled by a particular node (or it reaches a leaf and is not used     * in lower levels).     */    // FIFO queue of nodes to train: (treeIndex, node)    val nodeQueue = new mutable.Queue[(Int, Node)]()    val rng = new scala.util.Random()    rng.setSeed(seed)    // Allocate and queue root nodes.    val topNodes: Array[Node] = Array.fill[Node](numTrees)(Node.emptyNode(nodeIndex = 1))    Range(0, numTrees).foreach(treeIndex => nodeQueue.enqueue((treeIndex, topNodes(treeIndex))))    while (nodeQueue.nonEmpty) {      // Collect some nodes to split, and choose features for each node (if subsampling).      // Each group of nodes may come from one or multiple trees, and at multiple levels.      val (nodesForGroup, treeToNodeToIndexInfo) =        RandomForest.selectNodesToSplit(nodeQueue, maxMemoryUsage, metadata, rng) // 对decision tree没有意义，nodeQueue只有一个node，不需要选      // 4. Choose node splits, and enqueue new nodes as needed.      DecisionTree.findBestSplits(baggedInput, metadata, topNodes, nodesForGroup,        treeToNodeToIndexInfo, splits, bins, nodeQueue, timer)    }    val trees = topNodes.map(topNode => new DecisionTreeModel(topNode, strategy.algo))    RandomForestModel.build(trees)  }

//bins数目最大不能超过训练集中样本的size    val maxPossibleBins = math.min(strategy.maxBins, numExamples).toInt    //设置默认值    val numBins = Array.fill[Int](numFeatures)(maxPossibleBins)    if (numClasses > 2) {      // Multiclass classification      val maxCategoriesForUnorderedFeature =        ((math.log(maxPossibleBins / 2 + 1) / math.log(2.0)) + 1).floor.toInt      strategy.categoricalFeaturesInfo.foreach { case (featureIndex, numCategories) =>        // Decide if some categorical features should be treated as unordered features,        //  which require 2 * ((1 << numCategories - 1) - 1) bins.        // We do this check with log values to prevent overflows in case numCategories is large.        // The next check is equivalent to: 2 * ((1 << numCategories - 1) - 1) <= maxBins        if (numCategories <= maxCategoriesForUnorderedFeature) {          unorderedFeatures.add(featureIndex)          numBins(featureIndex) = numUnorderedBins(numCategories)        } else {          numBins(featureIndex) = numCategories        }      }    } else {      // Binary classification or regression      strategy.categoricalFeaturesInfo.foreach { case (featureIndex, numCategories) =>        numBins(featureIndex) = numCategories      }    }

/**   * Given the arity of a categorical feature (arity = number of categories),   * return the number of bins for the feature if it is to be treated as an unordered feature.   * There is 1 split for every partitioning of categories into 2 disjoint, non-empty sets;   * there are math.pow(2, arity - 1) - 1 such splits.   * Each split has 2 corresponding bins.   */  def numUnorderedBins(arity: Int): Int = 2 * ((1 << arity - 1) - 1)

/**   * Number of splits for the given feature.   * For unordered features, there are 2 bins per split.   * For ordered features, there is 1 more bin than split.   */  def numSplits(featureIndex: Int): Int = if (isUnordered(featureIndex)) {    numBins(featureIndex) >> 1  } else {    numBins(featureIndex) - 1  }

/**   * Returns splits and bins for decision tree calculation.   * Continuous and categorical features are handled differently.   *   * Continuous features:   *   For each feature, there are numBins - 1 possible splits representing the possible binary   *   decisions at each node in the tree.   *   This finds locations (feature values) for splits using a subsample of the data.   *   * Categorical features:   *   For each feature, there is 1 bin per split.   *   Splits and bins are handled in 2 ways:   *   (a) "unordered features"   *       For multiclass classification with a low-arity feature   *       (i.e., if isMulticlass && isSpaceSufficientForAllCategoricalSplits),   *       the feature is split based on subsets of categories.   *   (b) "ordered features"   *       For regression and binary classification,   *       and for multiclass classification with a high-arity feature,   *       there is one bin per category.   *   * @param input Training data: RDD of [[org.apache.spark.mllib.regression.LabeledPoint]]   * @param metadata Learning and dataset metadata   * @return A tuple of (splits, bins).   *         Splits is an Array of [[org.apache.spark.mllib.tree.model.Split]]   *          of size (numFeatures, numSplits).   *         Bins is an Array of [[org.apache.spark.mllib.tree.model.Bin]]   *          of size (numFeatures, numBins).   */  protected[tree] def findSplitsBins(      input: RDD[LabeledPoint],      metadata: DecisionTreeMetadata): (Array[Array[Split]], Array[Array[Bin]]) = {    val numFeatures = metadata.numFeatures    // Sample the input only if there are continuous features.    val hasContinuousFeatures = Range(0, numFeatures).exists(metadata.isContinuous)    val sampledInput = if (hasContinuousFeatures) {  // 对于连续特征，取值会比较多，需要做抽样      // Calculate the number of samples for approximate quantile calculation.      val requiredSamples = math.max(metadata.maxBins * metadata.maxBins, 10000) // 抽样数要远大于桶数      val fraction = if (requiredSamples < metadata.numExamples) { // 设置抽样比例        requiredSamples.toDouble / metadata.numExamples      } else {        1.0      }      input.sample(withReplacement = false, fraction, new XORShiftRandom().nextInt()).collect()    } else {      new Array[LabeledPoint](0)    }    metadata.quantileStrategy match {      case Sort =>        val splits = new Array[Array[Split]](numFeatures) // 初始化splits和bins         val bins = new Array[Array[Bin]](numFeatures)        // Find all splits.        // Iterate over all features.        var featureIndex = 0        while (featureIndex < numFeatures) { // 遍历所有的feature          val numSplits = metadata.numSplits(featureIndex) // 取出前面算出的splits和bins的数目          val numBins = metadata.numBins(featureIndex)          if (metadata.isContinuous(featureIndex)) { // 对于连续的feature            val numSamples = sampledInput.length            splits(featureIndex) = new Array[Split](numSplits)            bins(featureIndex) = new Array[Bin](numBins)            val featureSamples = sampledInput.map(lp => lp.features(featureIndex)).sorted // 从sampledInput里面取出该feature的所有取值，排序            val stride: Double = numSamples.toDouble / metadata.numBins(featureIndex) // 取样数/桶数，决定split(划分)的步长            logDebug("stride = " + stride)            for (splitIndex <- 0 until numSplits) { // 开始划分              val sampleIndex = splitIndex * stride.toInt // 划分数×步长，得到划分所用的sample的index              // Set threshold halfway in between 2 samples.              val threshold = (featureSamples(sampleIndex) + featureSamples(sampleIndex + 1)) / 2.0 // 划分点选取在前后两个sample的均值              splits(featureIndex)(splitIndex) =                new Split(featureIndex, threshold, Continuous, List()) // 创建Split对象            }            bins(featureIndex)(0) = new Bin(new DummyLowSplit(featureIndex, Continuous), // 初始化第一个split，DummyLowSplit，取值是Double.MinValue              splits(featureIndex)(0), Continuous, Double.MinValue)            for (splitIndex <- 1 until numSplits) { // 创建所有的bins               bins(featureIndex)(splitIndex) =                 new Bin(splits(featureIndex)(splitIndex - 1), splits(featureIndex)(splitIndex),                  Continuous, Double.MinValue)            }            bins(featureIndex)(numSplits) = new Bin(splits(featureIndex)(numSplits - 1), // 初始化最后一个split，DummyHighSplit，取值是Double.MaxValue              new DummyHighSplit(featureIndex, Continuous), Continuous, Double.MinValue)          } else { // 对于分类的feature             // Categorical feature            val featureArity = metadata.featureArity(featureIndex) // 离散特征中的取值个数            if (metadata.isUnordered(featureIndex)) { // 无序的离散特征              // TODO: The second half of the bins are unused.  Actually, we could just use              //       splits and not build bins for unordered features.  That should be part of              //       a later PR since it will require changing other code (using splits instead              //       of bins in a few places).              // Unordered features              //   2^(maxFeatureValue - 1) - 1 combinations              splits(featureIndex) = new Array[Split](numSplits)              bins(featureIndex) = new Array[Bin](numBins)              var splitIndex = 0              while (splitIndex < numSplits) {                val categories: List[Double] =                  extractMultiClassCategories(splitIndex + 1, featureArity)                splits(featureIndex)(splitIndex) =                  new Split(featureIndex, Double.MinValue, Categorical, categories)                bins(featureIndex)(splitIndex) = {                  if (splitIndex == 0) {                    new Bin(                      new DummyCategoricalSplit(featureIndex, Categorical),                      splits(featureIndex)(0),                      Categorical,                      Double.MinValue)                  } else {                    new Bin(                      splits(featureIndex)(splitIndex - 1),                      splits(featureIndex)(splitIndex),                      Categorical,                      Double.MinValue)                  }                }                splitIndex += 1              }            } else { // 有序的离散特征，不需要事先算，因为splits就等于featureArity               // Ordered features              //   Bins correspond to feature values, so we do not need to compute splits or bins              //   beforehand.  Splits are constructed as needed during training.              splits(featureIndex) = new Array[Split](0)              bins(featureIndex) = new Array[Bin](0)            }          }          featureIndex += 1        }        (splits, bins)      case MinMax =>        throw new UnsupportedOperationException("minmax not supported yet.")      case ApproxHist =>        throw new UnsupportedOperationException("approximate histogram not supported yet.")    }  }

private def labeledPointToTreePoint(      labeledPoint: LabeledPoint,      bins: Array[Array[Bin]],      featureArity: Array[Int],      isUnordered: Array[Boolean]): TreePoint = {    val numFeatures = labeledPoint.features.size    val arr = new Array[Int](numFeatures)    var featureIndex = 0    while (featureIndex < numFeatures) {      arr(featureIndex) = findBin(featureIndex, labeledPoint, featureArity(featureIndex),        isUnordered(featureIndex), bins)      featureIndex += 1    }    new TreePoint(labeledPoint.label, arr)  //只是将labeledPoint中的value替换成arr  }

// find best split for each node          val (split: Split, stats: InformationGainStats, predict: Predict) =            binsToBestSplit(aggStats, splits, featuresForNode, nodes(nodeIndex))          (nodeIndex, (split, stats, predict))        }.collectAsMap()

private def binsToBestSplit(      binAggregates: DTStatsAggregator,      splits: Array[Array[Split]],      featuresForNode: Option[Array[Int]],      node: Node): (Split, InformationGainStats, Predict) = {    // For each (feature, split), calculate the gain, and select the best (feature, split).    val (bestSplit, bestSplitStats) =      Range(0, binAggregates.metadata.numFeaturesPerNode).map { featureIndexIdx =>  //遍历每个feature       //......取出feature对应的splits           // Find best split.        val (bestFeatureSplitIndex, bestFeatureGainStats) =          Range(0, numSplits).map { case splitIdx =>    //遍历每个splits            val leftChildStats = binAggregates.getImpurityCalculator(nodeFeatureOffset, splitIdx)            val rightChildStats = binAggregates.getImpurityCalculator(nodeFeatureOffset, numSplits)            rightChildStats.subtract(leftChildStats)            predictWithImpurity = Some(predictWithImpurity.getOrElse(              calculatePredictImpurity(leftChildStats, rightChildStats)))            val gainStats = calculateGainForSplit(leftChildStats,    //算出gain，InformationGainStats对象              rightChildStats, binAggregates.metadata, predictWithImpurity.get._2)            (splitIdx, gainStats)          }.maxBy(_._2.gain)    //找到gain最大的split的index         (splits(featureIndex)(bestFeatureSplitIndex), bestFeatureGainStats)      }      //......省略离散特征的case    }.maxBy(_._2.gain) //找到gain最大的feature的split     (bestSplit, bestSplitStats, predictWithImpurity.get._1)  }

private def calculatePredictImpurity(      leftImpurityCalculator: ImpurityCalculator,      rightImpurityCalculator: ImpurityCalculator): (Predict, Double) =  {    val parentNodeAgg = leftImpurityCalculator.copy    parentNodeAgg.add(rightImpurityCalculator)    val predict = calculatePredict(parentNodeAgg)    val impurity = parentNodeAgg.calculate()    (predict, impurity)  }

private def calculatePredict(impurityCalculator: ImpurityCalculator): Predict = {    val predict = impurityCalculator.predict    val prob = impurityCalculator.prob(predict)    new Predict(predict, prob)  }

/**   * Calculate the impurity from the stored sufficient statistics.   */  def calculate(): Double = Variance.calculate(stats(0), stats(1), stats(2))

@DeveloperApi  override def calculate(count: Double, sum: Double, sumSquares: Double): Double = {    if (count == 0) {      return 0    }    val squaredLoss = sumSquares - (sum * sum) / count    squaredLoss / count  }

/**   * Prediction which should be made based on the sufficient statistics.   */  def predict: Double = if (count == 0) {    0  } else {    stats(1) / count  }

2014-12-08