title: Why this LR code run on spark too slowly?
body: <p>because the MLlib not support the sparse input. So I run the flowing code, which support the sparse input format, on spark clusters. 
  And the setting is:</p>
  
  <ol>
  <li>5 nodes, each node with 8 cores(all the cpu on each node are 100%,
  98% for user model, when run the code). </li>
  <li>the input: 10,000,000+ instance, and 600,000+ dimension on HDFS</li>
  </ol>
  
  <p>The code is:</p>
  
  <pre><code>import java.util.Random
  import scala.collection.mutable.HashMap
  import scala.io.Source
  import org.apache.spark.SparkContext
  import org.apache.spark.rdd.RDD
  import org.apache.spark.util.Vector
  import java.lang.Math
  import org.apache.spark.broadcast.Broadcast
  
  object SparseLR {
    val lableNum = 1
    val dimNum = 632918 
    val iteration = 10
    val alpha = 0.1
    val lambda = 0.1
    val rand = new Random(42)
    var w = Vector(dimNum, _=&gt; rand.nextDouble)
  
    class SparserVector {
      var elements = new HashMap[Int, Double]
  
      def insert(index: Int, value: Double){
        elements += index -&gt; value;
      }
  
  
      def *(scale: Double): Vector = {
        var x = new Array[Double](dimNum)
        elements.keySet.foreach(k =&gt; x(k) = scale * elements.get(k).get)
        Vector(x)
      }
    }
    case class DataPoint(x: SparserVector, y: Int)
  
    def parsePoint(line: String): DataPoint = {
      var features = new SparserVector
      val fields = line.split("\t")
      //println("fields:" + fields(0))
      val y = fields(0).toInt
      fields.filter(_.contains(":")).foreach( f =&gt; {
        val feature = f.split(":")
        features.insert(feature(0).toInt, feature(1).toDouble)
      })
      return DataPoint(features, y)
    }
  
    def gradient(p: DataPoint, w: Broadcast[Vector]) : Vector = {
      def h(w: Broadcast[Vector], x: SparserVector): Double = {
        val wb = w.value
        val features = x.elements
        val s = features.keySet.map(k =&gt; features.get(k).get * wb(k)).reduce(_ + _)
        1 / (1 + Math.exp(-p.y * s))
      }
      p.x * (-(1 - p.y *h(w, p.x)))
    }
  
    def train(sc: SparkContext, dataPoints: RDD[DataPoint]) {
      //val sampleNum = dataPoints.count
      val sampleNum = 11680250
  
      for(i &lt;- 0 until iteration) {
        val wb = sc.broadcast(w)
        val g = (dataPoints.map(p =&gt; gradient(p, wb)).reduce(_ + _) + lambda * wb.value) /sampleNum
        w -= alpha * g
  
        println("iteration " + i + ": g = " + g)
      }
    }
  
    def main(args : Array[String]): Unit = {
      System.setProperty("spark.executor.memory", "15g")
      System.setProperty("spark.default.parallelism", "32");
      val sc = new SparkContext("spark://xxx:12036", "LR", "/xxx/spark", List("xxx_2.9.3-1.0.jar"))
      val lines = sc.textFile("hdfs:xxx/xxx.txt", 32)
  
      val trainset = lines.map(parsePoint _).cache()
  
      train(sc, trainset)
    }
  }
  </code></pre>
  
  <p>Can anyone help me? Thanks!</p>