scala - 这个 Spark/Scala 代码的性能瓶颈在哪里？

Question

首先让我指出，我对 Spark 和 Scala 都很陌生。我一直试图通过尝试迁移我过去完成的 Hadoop Map/Reduce 作业之一来调查 promise 的 Spark 性能。这项工作在 Hadoop 上需要 14 分钟，使用 3 台 r3.2xlarge 机器输入 16 个 170mb 的压缩 bzip 文件。我尽我所能将它翻译成 Scala/Spark 成这样:

val conceptData = spark.textFile(inputPath)
val result = conceptData.repartition(60).cache()
    .map(line => {val metrics = JsonUtil.fromJson[ArticleMetrics](line); (metrics.source, metrics.data.get("entities").get)})
    .flatMap(metrics => metrics._2.map(t => (t._1,(1,List((metrics._1,t._2.head))))))
    .reduceByKey((a,b) => combine(a,b))
    .map(t => t._1 + "\t" + t._2._1 + "\t" + print(t._2._2))
result.saveAsTextFile(outputPath)

def print(tuples: List[(String, Any)]): String =
{
  tuples.map(l => l._1 + "\u200e" + l._2).reduce(_ + "\u200f" + _)
}

def combine(a: (Int, List[(String, Any)]), b: (Int, List[(String, Any)])): (Int, List[(String, Any)]) =
{
  (a._1 + b._1,a._2 ++ b._2)
}

object JsonUtil {
  val mapper = new ObjectMapper() with ScalaObjectMapper
  mapper.registerModule(DefaultScalaModule)
  mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)

  def fromJson[T](json: String)(implicit m : Manifest[T]): T = {
    mapper.readValue[T](json)
  }
}

我在开始时使用 repartition 命令将分区设置为 60，因为我读到了每个内核有 2-3 个分区的好地方。
我在相同的 3x r3.2xlarge 机器上运行这个 Spark 作业(每台机器有 8 个内核和 58G 可用)，所以我按以下方式提交我的作业:

spark/bin/spark-submit --executor-memory 58G --total-executor-cores 24 (... other arguments ...)

运行相同的输入需要 1 个多小时......我不确定问题是否出在 Scala 或 Spark 配置中，因此欢迎提供任何帮助。

此致，
奥古斯托

编辑 1:某些操作的平均时间:

从 S3 读取文件:~ 2 分钟

flatMap:~ 11 分钟

reduceByKey:> 1 小时

使用的 key 是 S3 路径，因此它们可能会很长，不知道这是否有所不同。

编辑 2:我替换了 reduceByKey功能与 .reduceByKey((a,b) => a)并且工作在 10 分钟内结束，所以 combine 肯定有问题功能

Answer 1

基于大部分时间都花在 flatMap 之后的事实，我怀疑 shuffle 会降低你的速度，而不是 CPU 利用率。您可能想尝试使用较少的分区运行作业。您可以尝试的另一件事是更换 reduceByKey()与 foldByKey()这是关联的但不是可交换的，这意味着它在运行组合时必须保持 RDD 的顺序，这可能会转化为在 shuffle 期间减少网络流量。