Apache Spark Java 8中的自连接示例

Question

我有一个如下的数据集

Kolkata,30
Delhi,23
Lucknow,33
Lucknow,36
Delhi,31
Kolkata,34
Delhi,21
Kolkata,23

我想做一个自连接，以获得表单的结果集。

Lucknow -> (30, 36), Kolkata -> (30, 34), (34, 23), (23, 30).

如何使用Spark RDD实现这一点。

JavaPairRDD<String, Integer> words = file.mapToPair(s ->{
    String[] temp = s.split(",");
    return new Tuple2<String, Integer>(temp[0], new Integer(temp[1]));
});        

JavaPairRDD<String, Iterable<Integer>> temp1 = words.groupByKey();        
JavaPairRDD<String, Iterable<Integer>> temp2 = words.groupByKey();            
JavaPairRDD<String, Tuple2<Iterable<Integer>, Iterable<Integer>>> words3 = temp2.join(temp1);

现在如何在元组上迭代？

Answer 1

要获取<String, Iterable<Int,Int>>的RDD，请在join之后使用.groupByKey()。

JavaPairRDD<String, Iterable<Tuple2<Integer>>> result = 
      words.join(words).groupByKey()

我可能把Java类型弄错了，但这是正确的操作顺序。我更习惯使用Python和Scala，因为它们不需要指定结果RDDs的类型。这些其他语言中的Spark调用执行相同的数据操作。

注：groupByKey() is known for being slow and resource intensive on large datasets

在word.join(words)之后，您应该有一个包含以下元素的RDD：

 (Kolkata, (30,34))
 (Kolkata, (30,23))
 ...

.groupByKey()通过键将所有值组合在一起，这样每个键只有一行，如下所示：

 (Kolkata, [(30,34), (30,23), (34,23), ...])

Answer 2

对于每个唯一的键，rdd self join会产生值的排列对。

要删除重复的值对(考虑到您的数据没有唯一键的重复值)，filterduplicates()将从排列集中删除那些重复的值对。

//outside main function
type pair=(String,(Int,Int))

def filterduplicates(p:pair):Boolean={
return p._2._1<p._2._2
}    

//insidemain
val rdd=sc.textFile("../cities.txt")
val mapped=rdd.map(l=>l.split(",")).map(l=>(l(0),l(1).toInt))

val joined=mapped.join(mapped) 
//(Lucknow,(33,33)) (Lucknow,(33,36)) (Lucknow,(36,33)) (Lucknow,(36,36))

val grouped=joined.filter(filterduplicates)
//(Lucknow,(33,36))

val listt=grouped.groupByKey().mapValues(_.toList)
val finalresult=listt.collect()
finalresult.foreach(println)
//(Delhi,List((23,31), (21,23), (21,31)))(Kolkata,List((30,34),(23,30)  (23,34))) (Lucknow,List((33,36)))