Оптимизация задач Spark

Question

Я пытаюсь найти оптимизированный способ генерации списка уникальных пар совпадений. Я попытался сделать это, используя серию плоских карт и различных запросов, но я обнаружил, что плоская карта не слишком сильно работает при работе над миллионами записей. Любая помощь в оптимизации этого будет получена с благодарностью.

Набор данных (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")) 

Answer 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)} 

Answer 2

Посмотрите на декартово функцию. Он создает RDD, что является всеми возможными комбинациями входных RDD. Обратите внимание, что это дорогостоящая операция (N^2 в размере РДУ)

Cartesian example