为什么这段LR代码在Spark上运行如此缓慢？

Question

为什么这段LR代码在Spark上运行如此缓慢？

4

因为MLlib不支持稀疏输入，所以我在Spark集群上运行了支持稀疏输入格式的下面代码。设置如下：

5个节点，每个节点有8个核心（每个节点的所有CPU都是100%，当运行代码时，用户模型占98%）。
输入：存储在HDFS中的10,000,000+个实例和600,000+维度。

代码如下：

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, _=> rand.nextDouble)

  class SparserVector {
    var elements = new HashMap[Int, Double]

    def insert(index: Int, value: Double){
      elements += index -> value;
    }


    def *(scale: Double): Vector = {
      var x = new Array[Double](dimNum)
      elements.keySet.foreach(k => 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 => {
      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 => 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 <- 0 until iteration) {
      val wb = sc.broadcast(w)
      val g = (dataPoints.map(p => 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)
  }
}

有人能帮我吗？谢谢！

- Tim

很难说，你能将分析器钩入任务中吗？ - Thomas Jungblut

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Rüdiger Klaehn · Accepted Answer

这真的很难给出一个确切答案。也许这更适合于代码审查stackoverflow子站点？

一些立即显而易见的事情：您的渐变函数似乎效率低下。当您想要对映射的每个键/值对执行某些操作时，最好使用

for((k,v)<-map) { 
  ...
}

比做更好

for(k<-map.keySet) { val value = map.get(k).get; 
  ... 
}

此外，对于像这样的性能关键代码，更好的选择是将 reduce 改为累加可变值。因此，重新编写的梯度函数如下所示。

def gradient(p: DataPoint, w: Broadcast[Vector]) : Vector = {
  def h(w: Broadcast[Vector], x: SparserVector): Double = {
    val wb = w.value
    val features = x.elements
    var s = 0.0
    for((k,v)<-features)
      s += v * wb(k)
    1 / (1 + Math.exp(-p.y * s))
  }
  p.x * (-(1 - p.y *h(w, p.x)))
}

现在，如果您想进一步提高性能，您需要将SparseVector更改为使用索引数组和值数组而不是Map [Int，Double]。原因是，在Map中，键和值将作为对象进行装箱并具有相当大的开销，而Array [Int]或Array [Double]仅是单个紧凑的内存块。

（为方便起见，建议定义一个构建器，该构建器使用SortedMap [Int，Double]并在完成构建时转换为两个数组）

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

  def *(scale: Double): Vector = {
    var x = new Array[Double](dimNum)
    var i = 0
    while(i < indices.length) {  
      x(indices(i)) = scale * values(i) 
      i += 1
    }
    Vector(x)
  }
}

请注意，上面的代码示例没有经过测试，但我想你会明白的。