將RDD的每個元素添加到Spark Scala中另一個RDD的每個元素。

Question

我有兩個與RDDS以下結構

org.apache.spark.rdd.RDD[(Long, Double)] 

這裏RDD的每一行包含一個索引Long和值Double。我想將RDD的每個元素添加到Spark Scala中其他RDD的每個元素。

一個例子是這樣的：

RDD1集：

Array[(Long, Double)] = Array((0,-3),(1,2)) 

RDD2：

Array[(Long, Double)] = Array((0,4),(1,-2)) 

結果：

Array[(Long, Double)] = Array((0,1),(0,-5),(1,6),(1,0)) 

Answer 1

你真正在這裏做什麼是你的兩個rdd的的笛卡爾乘積，在那裏你只需求和值每導致((key, value), (key, value))對，保持第一元組的關鍵：

val result = rdd1.cartesian(rdd2).map(x => (x._1._1, x._2._2 + x._1._2)) 
// Result 
result.collect() 
Array[(Int, Int)] = Array((0,1), (0,-5), (1,6), (1,0)) 

小心使用cartesian()不過，內存消耗將大大增加你的rdd的更大。

Answer 2

請試試這個：

val df1 = Seq((0,-3),(1,2)).toDF("col1", "col2") 
val df2 = Seq((0,4),(1,-2)).toDF("col1", "col2") 

df1.createOrReplaceTempView("temp1") 
df2.createOrReplaceTempView("temp2") 

spark.sql("SELECT t1.col1 + t2.col1, t1.col2 + t2.col2 FROM t1, t2").show 

Answer 3

拉鍊兩個RDDS然後映射在該計算的款項