根據scala中的密鑰比較兩對RDD的值

Question

有沒有一種方法可以根據密鑰比較2對RDDs的值。
例如：

RDD1: RDD[(Int, String)] = {(1, "ABC"), (2, "XYZ"), (3, "PQR")} 
RDD2: RDD[(Int, String)] = {(5, "AAA"), (2, "XYZ"), (3, "LMN")} 

我的任務是讓存在於rdds其中值是不同的密鑰數量。
該連接給出了rdd以及鍵2和3。 但我想最終輸出爲3因爲值不同。

我已經嘗試下面的方法：

var diff = rdd1.join(rdd2).map{case(k,(s1,s2))=> if (s1!=s2) (k,s1)} 

這樣我就可以得到差異的計數。但它提供了兩個鍵2和3。我想我需要其他條件跳過匹配值。
任何人都可以請建議我如何能達到預期的效果嗎？ （或使用任何不同的方法）。

Answer 1

使用collect，它採用部分函數並刪除未定義部分函數的值。隨後，包括「如果」作爲後衛在case語句，而不是案件的結果表達式的一部分：

val diff = rdd1.join(rdd2).collect { 
    case (k, (s1, s2)) if s1 != s2 => (k, s1) 
} 

這樣，部分功能將不會被定義爲其中s1 == s2，這將使collect將它們濾除。

的等效（但略低於優雅）解決方案將使用map和filter - 單獨map總是返回相同數量的記錄作爲輸入：

val diff = rdd1.join(rdd2).filter { 
    case (k, (s1, s2)) => s1 != s2 
}.map { 
    case (k, (s1, s2)) => (k, s1) 
} 

Answer 2

使用leftInnerJoin到關鍵存在於第一RDD和依靠它。

val diff = rdd1.leftOuterJoin(rdd2).count()