Restored broadcasting

mllib/src/main/scala/org/apache/spark/mllib/optimization/GradientDescent.scala

@@ -182,18 +182,17 @@ object GradientDescent extends Logging {
       weights, Vectors.dense(new Array[Double](weights.size)), 0, 1, regParam)._2
 
     for (i <- 1 to numIterations) {
+      val bcWeights = data.context.broadcast(weights)
       // Sample a subset (fraction miniBatchFraction) of the total data
       // compute and sum up the subgradients on this subset (this is one map-reduce)
       val (gradientSum, lossSum) = data.sample(false, miniBatchFraction, 42 + i)
-        .treeAggregate((BDV.zeros[Double](weights.size), 0.0))(
-          seqOp = (c, v) => (c, v) match {
-            case ((grad, loss), (label, features)) =>
-              val l = gradient.compute(features, label, weights, Vectors.fromBreeze(grad))
-              (grad, loss + l)
+        .treeAggregate((BDV.zeros[Double](n), 0.0))(
+          seqOp = (c, v) => (c, v) match { case ((grad, loss), (label, features)) =>
+            val l = gradient.compute(features, label, bcWeights.value, Vectors.fromBreeze(grad))
+            (grad, loss + l)
           },
-          combOp = (c1, c2) => (c1, c2) match {
-            case ((grad1, loss1), (grad2, loss2)) =>
-              (grad1 += grad2, loss1 + loss2)
+          combOp = (c1, c2) => (c1, c2) match { case ((grad1, loss1), (grad2, loss2)) =>
+            (grad1 += grad2, loss1 + loss2)
           })
 
       /**