Spark 性能 - 如何并行化大循环？

Question

我有一个包含 8000 个循环的 Spark 应用程序，它在 5 个节点的集群上运行。每个节点有 125GB 内存和 32 个内核。相关代码如下所示：

for (m <- 0 until deviceArray.size) { // there are 1000 device 
  var id = deviceArray(m)

  for (t <- 1 to timePatterns) { // there are 8 time patterns
     var hrpvData = get24HoursPVF(dataDF, id, t).cache()

  var hrpvDataZI = hrpvData.zipWithIndex

  var clustersLSD = runKMeans(hrpvData, numClusters, numIterations)

  var clusterPVPred = hrpvData.map(x => clustersLSD.predict(x))
  var clusterPVMap = hrpvDataZI.zip(clusterPVPred)

  var pvhgmRDD = clusterPVMap.map{r => (r._2, r._1._2)}.groupByKey

  var arrHGinfo = pvhgmRDD.collect 

  // Post process data 
  // .....

  hrpvData.unpersist()
  }
}

函数调用get24HoursPVF()为k-means准备特征向量，大约需要40秒。每个循环大约需要 50 秒才能完成使用集群。我的数据大小为 2 到 3 GB（从表中读取）。给定 8000 个循环，运行此 Spark 应用程序的总时间是不可接受的 (8000x50s)。

由于每个设备都是独立的，有没有办法并行化 8000 次迭代？或者如何利用集群来解决总运行时间长的问题？ Scala Future 不会工作，因为它只是同时提交作业，但 Spark 不会同时运行这些作业。

Answer 1

除了 for 循环之外，您的代码中有两个 Spark 中最慢的 API 调用 - groupByKey 和 collect。

groupByKey 几乎不应该使用，而是查看reduceByKey，查看此Databricks blog 了解更多详细信息。

collect 将该 RDD 中的所有数据传输到驱动节点上的一个数组中，除非这是少量数据，否则会对性能产生相当大的影响。

在 for 循环中，我不是特别熟悉您要执行的操作，但在

var hrpvData = get24HoursPVF(dataDF, id, t).cache()

您正在为每个 id 和 t 值构建和缓存一个新数据帧。我不知道为什么你不能在一开始就构建一个包含每个 id 和 t 变体的数据帧，然后在整个数据帧上运行你的 zipWithIndex、map 等？