spark.mllib 中的并行度

Question

假设我有一个 data 类型的对象 Array[RDD]。我想在此对象中的每个 RDD 上学习 independent 机器学习模型。例如，对于随机森林：

data.map{ d => RandomForest.trainRegressor(d,2,Map[Int,Int](),2,"auto","gini",2,10) }

当我用 spark-submit --master yarn-client ... 启动这个作业时，独立学习任务似乎没有在多个节点上并行化。几乎所有的工作都是由一个节点（这里是节点 10）完成的，从应用程序的屏幕截图可以看出 UI:

附录

为完整起见，完整代码如下：

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.tree.RandomForest


object test {
  def main(args: Array[String]) {

    val conf = new SparkConf().setMaster("local").setAppName("test")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)

    // Load data
    val rawData = sc.textFile("data/mllib/sample_tree_data.csv")
    val data = rawData.map { line =>
      val parts = line.split(',').map(_.toDouble)
      LabeledPoint(parts(0), Vectors.dense(parts.tail))
    }

    val CV_data = (1 to 100).toArray.map(_ => {val splits = data.randomSplit(Array(0.7, 0.3)) ;  splits(0)})

    CV_data.map(d => RandomForest.trainClassifier(d, 2, Map[Int, Int](), 2, "sqrt", "gini", 2, 100))

    sc.stop()
    System.exit(0)
  }
}

Answer 1

问题是RandomForest.trainClassifier可以看作是一个action，因为它急切地触发了一些涉及到的RDD计算的执行。因此，每当您调用 RandomForest.trainClassifier 时，Spark 作业都会被提交到集群并执行。

由于 Scala Array 上的 map 操作是按顺序执行的，因此您最终会一个接一个地执行 trainClassifier 作业。为了并行执行作业，您必须在并行集合上调用 map。以下代码片段应该可以解决问题：

CV_data.par.map(d => RandomForest.trainClassifier(d, 2, Map[Int, Int](), 2, "sqrt", "gini", 2, 100))

spark.mllib 中的并行度

Parallelism in spark.mllib

parallel-processing

scala

apache-spark

apache-spark-mllib