5
\$\begingroup\$

Because the MLlib does not support the sparse input, I ran the following code, which supports the sparse input format, on spark clusters. The settings are:

  • 5 nodes, each node with 8 cores (all the CPU on each node are 100%, 98% for user model, when running the code).
  • the input: 10,000,000+ instance, and 600,000+ dimension on HDFS

For the above settings, the code cost 20+ minutes for one iteration.

I have asked this question here, where the first version of the code can be found. And the first version cost 3 hours for one iteration. @Rüdiger Klaehn gave me some suggestions, and I updated the code. The time for one iteration take from 3 hours to 20 minutes. For my application, this is also unacceptable, so can any one give me some suggestions?

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
import org.apache.spark.serializer.KryoSerializer
import org.apache.spark.serializer.KryoRegistrator
import com.esotericsoftware.kryo._
import scala.collection.mutable.ArrayBuffer

object SparseLRV3 {
  val lableNum = 1
  val dimNum = 632918
  val iteration = 10
  val alpha = 0.1
  val lambda = 0.1
  val rand = new Random(42)
  var w = new Array[Double](dimNum)

  class LRRegistrator extends KryoRegistrator {
    override def registerClasses(kryo: Kryo) {
      kryo.register(classOf[SparseVector])
      kryo.register(classOf[DataPoint])
    }
  }

  class SparseVector(val indices: Array[Int], val values: Array[Double]) {
    require(indices.length == values.length)

    def map(f: Double => Double): SparseVector = {
      new SparseVector(indices, values.map(x => f(x)).toArray)
    }

    def +(that: SparseVector): SparseVector = {
      var tups = new HashMap[Int, Double]
      //load touples from this
      for (i <- 0 until indices.length)
        tups += indices(i) -> values(i)
      //load touples from that
      for (i <- 0 until that.indices.length) {
        val idx = that.indices(i)
        val v = that.values(i)

        if (tups.contains(idx))
          tups(idx) += v 
        else
          tups += idx -> v
      }
      new SparseVector(tups.keys.toArray, tups.values.toArray) //return a new sparse vector
      output
    }
  }

  case class DataPoint(x: SparseVector, y: Int)

  def parsePoint(line: String): DataPoint = {
    val fields = line.split("\t")
    val y = fields(0).toInt
    var i = 0
    val len = fields.length - 1 //get the length
    var indices = new Array[Int](len)
    var values = new Array[Double](len)
    fields.filter(_.contains(":")).foreach(f => {
        val feature = f.split(":")
        indices(i) = feature(0).toInt
        values(i) = feature(1).toDouble
        i = i + 1
      })
    DataPoint(new SparseVector(indices, values), y)
  }

  def gradient(p: DataPoint, w: Broadcast[Array[Double]]): SparseVector = {
    def h(w: Broadcast[Array[Double]], p: DataPoint): Double = {
      val wb = w.value
      var s = 0.0
      for (i <- 0 until p.x.indices.length)
        s += p.x.values(i) * wb(p.x.indices(i))
      1 / (1 + Math.exp(-p.y * s))
    }

    val scale = -(1 - p.y * h(w, p))
    p.x.map(v => v * scale)
  }

  def train(sc: SparkContext, dataPoints: RDD[DataPoint]) {
    for (i <- 0 until iteration) {
      val wb = sc.broadcast(w)
      val g = dataPoints.map(p => gradient(p, wb)).reduce(_ + _)

      for (i <- 0 until g.indices.length)
        w(g.indices(i)) += alpha * g.values(i)

      for (i <- 0 until w.length)
        w(i) += lambda * wb.value(i)
    }
  }

  def main(args: Array[String]): Unit = {
    System.setProperty("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    System.setProperty("spark.kryo.registrator", "LRRegistrator")
    System.setProperty("spark.executor.memory", "15g")
    System.setProperty("spark.storage.blockManagerHeartBeatMs", "300000")
    val sc = new SparkContext("spark://xxxx:12036", "LR", "xxxx", List("xxxx.jar"))
    val lines = sc.textFile("hdfs://xxx", 40)
    val trainset = lines.map(parsePoint).cache()

    train(sc, trainset)
  }
}

I update a new version of code, this version use the accumulator to avoid allocation new object when do the add operation. and the time cost from 20 min for one iteration to 2 s. But I am not sure if the logic is right:

import java.util.Random
import scala.collection.mutable.HashMap
import scala.collection.immutable.TreeMap
import scala.io.Source
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import java.lang.Math
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.Accumulator
import org.apache.spark.AccumulableParam
import org.apache.spark.serializer.KryoSerializer
import org.apache.spark.serializer.KryoRegistrator
import com.esotericsoftware.kryo._
import scala.collection.mutable.ArrayBuffer
import org.apache.spark.AccumulatorParam

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 = Array.tabulate(dimNum)(_ => rand.nextDouble)

  class LRRegistrator extends KryoRegistrator {
    override def registerClasses(kryo: Kryo) {
      kryo.register(classOf[SparseVector])
      kryo.register(classOf[DataPoint])
      kryo.register(classOf[SparseLR.Vector])
      kryo.register(classOf[Array[Double]])
    }
  }

  class SparseVector(val indices: Array[Int], val values: Array[Double]) {
    require(indices.length == values.length)

    def map(f: Double => Double): SparseVector = {
      new SparseVector(indices, values.map(x => f(x)).toArray)
    }
  }

  case class DataPoint(x: SparseVector, y: Int)

  def parsePoint(line: String) : DataPoint  = {
    val fields = line.split("\t")
    val y = fields(0).toInt
    var i = 0
    val len = fields.length - 1 //get the length
    var indices = new Array[Int](len)
    var values = new Array[Double](len)
    fields.filter(_.contains(":")).foreach(f => {
        val feature = f.split(":")
        indices(i) = feature(0).toInt
        values(i) = feature(1).toDouble
        i = i + 1
      })
    DataPoint(new SparseVector(indices, values), y)
  }


  class Vector(val data: Array[Double]) extends Serializable {

  }

  implicit object VectorAP extends AccumulableParam[Vector, SparseVector] {
    def zero(v: Vector) = new Vector(new Array(v.data.size))
    def addInPlace(v1: Vector, v2: Vector) : Vector = {
      var i = 0
      while ( i < v1.data.size) {
        v1.data(i) += v2.data(i)
        i += 1
      }
      return v1
    }

    def addAccumulator(v1: Vector, v2: SparseVector) : Vector = {
      var i = 0
      while ( i < v2.indices.length ) {
        v1.data(v2.indices(i)) += v2.values(i)
        i += 1
      }
      v1
    }
  }


  def train(sc: SparkContext, dataPoints: RDD[DataPoint]) {
    var g_a = Array.tabulate(dimNum)(_ => 0.0)
    for (j <- 0 until iteration) {
      var wb = sc.broadcast(w)
      var gx = sc.accumulable(new Vector(g_a))(VectorAP)
      dataPoints.foreach { p =>
            var s = 0.0
            var i = 0
            while  (i <  p.x.indices.length) {
                s += p.x.values(i) * wb.value(p.x.indices(i))
                i = i + 1
            }
            var res = 1 / (1 + Math.exp(-p.y * s))
            val scale = -(1 - p.y * res)
            gx += p.x.map(v => v * scale) // call the function: addAccumulator 
      }

      val g = gx.value.asInstanceOf[Vector] // call the function: addInPlace
      var i = 0
      while ( i < g.data.length) {
        w(i) = alpha * g.data(i) + lambda * wb.value(i)
        i += 1
      }
    }
  }

  def main(args: Array[String]): Unit = {
    System.setProperty("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    System.setProperty("spark.closure.serializer", "org.apache.spark.serializer.JavaSerializer")
    System.setProperty("spark.kryo.registrator", "LRRegistrator")
    System.setProperty("spark.executor.memory", "15g")
    System.setProperty("spark.default.parallelism", "48")
    System.setProperty("spark.storage.blockManagerHeartBeatMs", "300000")
    val sc = new SparkContext("spark://xxx:12036", "LR", "xxxx", List("xxxx.jar"))
    val lines = sc.textFile("hdfs://xxxx/", 40)
    val trainset = lines.map(parsePoint).cache()
    train(sc, trainset)
  }
}
\$\endgroup\$
  • \$\begingroup\$ If you'd like your added code to be reviewed, I recommend posting it as a new question with a link back to this question. \$\endgroup\$ – Jamal Jan 4 '14 at 3:21
2
\$\begingroup\$

It seems to me that if you have very sparse vectors, converting to a non-sparse vector will be very inefficient. E.g. you have a sparse vector with 4 non-zero elements and convert it into a flat vector with 1000000 elements, the array allocation will take the most time.

You can of course delay the conversion for as long as possible and work with sparse vectors all the time, but if apache spark does not support sparse vectors you will have to convert at some point, so you might have to search for some other toolkit.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.