如何通过 Spark MLlib 并行训练和预测？答案

【问题标题】：How to train and predict in parallel via Spark MLlib?如何通过 Spark MLlib 并行训练和预测？
【发布时间】：2016-05-29 13:02:02
【问题描述】：

我必须解决不同限制的回归问题。他们有一堆标准和规则，我必须建立模型并为每一个做出预测，结合所有并保存。所以，现在我的解决方案如下：

criteria2Rules: List[(String, Set[String])]

var result: RDD[(Id, Double)] = sc.parallelize(Array[(Id, Double)]())
criteria2Rules.foreach {
  case (criterion, rules) =>
    val trainDataSet: RDD[LabeledPoint] = prepareTrainSet(criterion, data)
    val model: GradientBoostedTreesModel = buildModel(trainDataSet)
    val predictionDataSet = preparePredictionDataSet(criterion, data)
    val predictedScores = predictScores(predictionDataSet, model, criterion, rules)
    result = result.union(predictedScores)
}

效果几乎不错，但是太慢了，因为 GradientBoostedTreesModel 训练的速度不是那么快，尤其是在特征量大、样本量大以及使用标准列表非常大的情况下。如果 Spark 能够并行训练模型并进行预测，我想它会更好地工作。

我尝试过使用关系型的数据操作方式：

val criteria2RulesRdd: RDD[(String, Set[String])]

val cartesianCriteriaRules2DataRdd = criteria2RulesRdd.cartesian(dataRdd)
cartesianCriteriaRules2DataRdd
  .aggregateByKey(List[Data]())(
    { case (lst, tuple) => lst :+ tuple }, { case (lstL, lstR) => lstL ::: lstR}
  )
  .map {
    case (criteria, rulesSet, scorePredictionDataList) =>
      val trainSet = ???
      val model = ???
      val predictionSet = ???
      val predictedScores = ???
  }
  ...

但是当一个 RDD 在另一个 RDD 中产生时，它不可避免地会带来这种情况（GradientBoostedTreesModel 是在 RDD[LabeledPoint] 上训练的），据我所知这是一个坏的场景。

有什么方法可以并行使用 Spark MLlib？

【问题讨论】：

标签： scala apache-spark apache-spark-mllib

【解决方案1】：

我在 Spark 用户邮件列表中问了同样的问题，得到了答案：

“如果你有一个大集群，你可以在驱动程序的不同线程中触发训练作业。将 RDD 放在一个 RDD 中是行不通的。简单的例子：

(0 until 10).par.foreach { i =>
  val n = sc.parallelize(0 until 1000000).count()
  println(s"$i: $n")
}

【讨论】：