Spark MLlib之线性回归源码分析

    LinearRegressionWithSGD伴生对象中train方法实现

  def train(
      input: RDD[LabeledPoint],
      numIterations: Int,
      stepSize: Double,//默认步长为1
      miniBatchFraction: Double)//每次跌打使用的batch因子，默认为1
    : LinearRegressionModel =
  {
    new LinearRegressionWithSGD(stepSize, numIterations, miniBatchFraction).run(input)
  }

   LinearRegressionWithSGD中run方法实现

  def run(input: RDD[LabeledPoint], initialWeights: Array[Double]) : M = {

    // Check the data properties before running the optimizer
    if (validateData && !validators.forall(func => func(input))) {//预验证输入数据的合法性，validators中存储验证的所有方法
      throw new SparkException("Input validation failed.")
    }
    // Prepend an extra variable consisting of all 1.0's for the intercept.
    val data = if (addIntercept) {//判读是否需要添加intercept
      input.map(labeledPoint => (labeledPoint.label, 1.0 +: labeledPoint.features))
    } else {
      input.map(labeledPoint => (labeledPoint.label, labeledPoint.features))
    }//对象转化为元组(类标签,特征)

    val initialWeightsWithIntercept = if (addIntercept) {
      0.0 +: initialWeights
    } else {
      initialWeights
    }//初始化权重特征

    val weightsWithIntercept = optimizer.optimize(data, initialWeightsWithIntercept)//返回优化后的权重

    val (intercept, weights) = if (addIntercept) {
      (weightsWithIntercept(0), weightsWithIntercept.tail)
    } else {
      (0.0, weightsWithIntercept)
    }

    logInfo("Final weights " + weights.mkString(","))
    logInfo("Final intercept " + intercept)

    createModel(weights, intercept)//使用计算后的权重向量和拦截参数创建模型
  }

其中optimizer.optimize(data, initialWeightsWithIntercept)是LinearRegressionWithSGD实现的核心，oprimizer的类型为GradientDescent，optimize方法中主要调用GradientDescent伴生对象的runMiniBatchSGD方法，返回当前迭代产生的最优特征权重向量。

GradientDescentd对象中optimize方法实现

  def optimize(data: RDD[(Double, Array[Double])], initialWeights: Array[Double])
    : Array[Double] = {
	//返回优化后的权重向量，和迭代过程中误差
    val (weights, stochasticLossHistory) = GradientDescent.runMiniBatchSGD(
        data,
        gradient,
        updater,
        stepSize,
        numIterations,
        regParam,
        miniBatchFraction,
        initialWeights)
    weights
  }

GradientDescent伴生对象中runMiniBatchSGD方法实现

  def runMiniBatchSGD(
    data: RDD[(Double, Array[Double])],
    gradient: Gradient,//SquaredGradient—平方剃度下降算法
    updater: Updater,//SimpleUpdater
    stepSize: Double,//1.0
    numIterations: Int,//100
    regParam: Double,//0.0
    miniBatchFraction: Double,//1.0
    initialWeights: Array[Double]) : (Array[Double], Array[Double]) = {

    val stochasticLossHistory = new ArrayBuffer[Double](numIterations)

    val nexamples: Long = data.count()
    val miniBatchSize = nexamples * miniBatchFraction

    // Initialize weights as a column vector／／创建一维向量，第一个参数为行数，第二个参数为列数，第三个参数开始为值
    var weights = new DoubleMatrix(initialWeights.length, 1, initialWeights:_*)
    var regVal = 0.0
    
    for (i <- 1 to numIterations) {
      /**
       * 使用平方梯度下降算法
       * gradientSum:本次选择迭代样本的梯度总和，
       * lossSum:本次选择迭代样本的误差总和
       */
      val (gradientSum, lossSum) = data.sample(false, miniBatchFraction, 42+i).map {
        case (y, features) =>//(标签，特征)
          val featuresCol = new DoubleMatrix(features.length, 1, features:_*)
          val (grad, loss) = gradient.compute(featuresCol, y, weights)//(特征，标签，特征属性权重向量)
          /**
           * class SquaredGradient extends Gradient {
			  override def compute(data: DoubleMatrix, label: Double, weights: DoubleMatrix): 
			      (DoubleMatrix, Double) = {
			    val diff: Double = data.dot(weights) - label//计算当前计算对象的类标签与实际类标签值之差
			    val loss = 0.5 * diff * diff//当前平方梯度下降值
			    val gradient = data.mul(diff)
			    (gradient, loss)
			  }
			}
           */
          (grad, loss)//当前训练对象的特征权重向量和误差
      }.reduce((a, b) => (a._1.addi(b._1), a._2 + b._2)) //计算本次迭代所选训练数据新的特征权重向量之和和误差总和     
      /**
       * NOTE(Xinghao): lossSum is computed using the weights from the previous iteration
       * and regVal is the regularization value computed in the previous iteration as well.
       */
      stochasticLossHistory.append(lossSum / miniBatchSize + regVal)//miniBatchSize=样本中对象数量＊batch因子，regVal为回归因子
      val update = updater.compute(
        weights, gradientSum.div(miniBatchSize), stepSize, i, regParam)//weights：属性向量中设置的权重因子，regParam：为回归参数，stepSize：计算步长，i：当前迭代次数
      /**
       * class SimpleUpdater extends Updater {
       * /**
          * weihtsOld:上一次迭代计算后的特征权重向量
          * gradient:本次迭代计算的特征权重向量
          * stepSize:迭代步长
          * iter:当前迭代次数
          * regParam:回归参数 
          */
		  override def compute(weightsOld: DoubleMatrix, gradient: DoubleMatrix,
		      stepSize: Double, iter: Int, regParam: Double): (DoubleMatrix, Double) = {
		    val thisIterStepSize = stepSize / math.sqrt(iter)//以当前迭代次数的平方根的倒数作为本次迭代趋近(下降)的因子
		    val normGradient = gradient.mul(thisIterStepSize)
		    (weightsOld.sub(normGradient), 0)//返回本次剃度下降后更新的特征权重向量
		  }
		}
	   *
       */
      weights = update._1
      regVal = update._2//使用SimpleUpdater值为0
    }

    logInfo("GradientDescent finished. Last 10 stochastic losses %s".format(
      stochasticLossHistory.takeRight(10).mkString(", ")))

    (weights.toArray, stochasticLossHistory.toArray)
  }

  在MiniBatchSGD中主要实现对输入数据集进行迭代抽样，通过使用SquaredGradient作为梯度下降算法，使用SimpleUpdater作为更新算法，不断对抽样数据集进行迭代计算从而找出最优的特征权重向量解。

  使用官方测试代码如下：

  def linearRegressionAPITest(sc: SparkContext) {
    val url = "/Users/yangguo/hadoop/spark/mllib/data/ridge-data/lpsa.data"
    val data = sc.textFile(url)
    val parseData = data.map { line =>
      val parts = line.split(',')
      LabeledPoint(parts(0).toDouble, parts(1).split(' ').map(x => x.toDouble).toArray)
    }
    val numIterations = 20
    val model = LinearRegressionWithSGD.train(parseData, numIterations)
    val valuesAndPreds = parseData.map { point =>
      val prediction = model.predict(point.features)
      (point, prediction)
    }
    valuesAndPreds.foreach { case (v, p) => print("[" + v.label + "," + p + "]"); v.features.foreach(base => print(base + "--")); println("\n") }
    val isSuccessed = valuesAndPreds.map { case (v, p) => math.pow((p - v.label), 2) }.reduce(_ + _) / valuesAndPreds.count
    println(isSuccessed)
  }

参考:

[1].http://rdc.taobao.org/?p=2163

[2].http://cs229.stanford.edu/notes/cs229-notes1.pdf

[3].http://blog.sina.com.cn/s/blog_62339a2401015jyq.html

[4].http://blog.csdn.net/pennyliang/article/details/6998517

[5].http://en.wikipedia.org/wiki/Lasso_(statistics)#Lasso_method