Spark任务优化答案

【问题标题】：Spark task optimisationSpark任务优化
【发布时间】：2016-04-20 12:09:16
【问题描述】：

我正在尝试找到一种优化的方法来生成独特的协同定位配对列表。我希望使用一系列平面图和不同的查询来做到这一点，但我发现平面图在运行数百万条记录时并没有过度表现。我们将不胜感激地收到任何优化此功能的帮助。

数据集是 (geohash, id)，我在 30 节点集群上运行它。

val rdd = sc.parallelize(Seq(("gh5", "id1"), ("gh4", "id1"), ("gh5", "id2"),("gh5", "id3"))

val uniquePairings = rdd.groupByKey().map(value =>
     value._2.toList.sorted.combinations(2).map{
     case Seq(x, y) => (x, y)}.filter(id => 
     id._1 != id._2)).flatMap(x => x).distinct()       

voutput = Array(("id1","id2"),("id1","id3"),("id2","id3"))

【问题讨论】：

标签： scala apache-spark optimization bigdata

【解决方案1】：

一个简单的join 在这里应该绰绰有余。例如DataFrames:

val df = rdd.toDF
df.as("df1").join(df.as("df2"),
  ($"df1._1" === $"df2._1") && 
  ($"df1._2" < $"df2._2")
).select($"df1._2", $"df2._2")

或数据集

val ds = rdd.toDS
ds.as("ds1").joinWith(ds.as("ds2"),
  ($"ds1._1" === $"ds2._1") && 
  ($"ds1._2" < $"ds2._2")
).map{ case ((_, x), (_, y)) => (x, y)}

【讨论】：

感谢您的回复，数据帧是否在改组方面提供了连接的任何性能优势。
我不知道，但它比 RDD 上的标准联合组 join 更有效。
谢谢，我运行了一个稍有改动的版本来首先对键进行排序，以便更有可能在本地进行查找。我认为 Spark 将首先在本地查找？

【解决方案2】：

查看笛卡尔函数。它产生一个 RDD，它是输入 RDD 的所有可能组合。请注意，这是一项昂贵的操作（RDD 大小为 N^2）

Cartesian example

【讨论】：

感谢您的回复，但使用笛卡尔运行相同的函数似乎要慢得多。我认为这是因为这不是真正的 n^2 问题。
当然可以。在笛卡尔运算之前尝试缓存 RDD 可能是值得的。再想一想，您的原始方法感觉像是完成此任务的最佳方法。但是，您应该使用 reduceByKey 而不是 groupByKey。这个任务是关联的（所以你不需要 groupByKey）并且 reduceByKey 将节省一些数据洗牌（databricks.gitbooks.io/databricks-spark-knowledge-base/content/…）
再次感谢我在完整示例中编写了 combineByKey，因为数据结构比示例稍微复杂一些，但感谢您的提醒。对于 1100 万条记录，这项任务似乎需要大约 4-5 个小时，这似乎有点偏高。不确定这是否现实。