スパークRDDの要素を互いに掛け合わせる

Question

Apache Sparkジョブを実行するときに直面した問題の1つは、RDDの各要素を互いに掛け合わせることです。 だけで、私はこれに似た何かをしたい、

を置く現在、私は、各「foreachの」の2つのイテレータを使用して、これをやっています。私の直感は、これが非常に効率的なやり方でできるということです。

for (elementOutSide <- iteratorA) { 
    for (elementInside <- iteratorB) { 
    if (!elementOutSide.get(3).equals(elementInside.get(3))) { 
     val multemp = elementInside.getLong(3) * elementOutSide.getLong(3) 
     .... 
     ... 

}}} 

状況を改善し改善するのにお手伝いできますか？前もって感謝します .. ！！

Answer 1

コメントで指摘されているように、これはデカルト結合です。ここでは、我々は2つずつの非同一Int Sの乗算に興味RDD[(Int, String)]、上で行うことができる方法は次のとおりです。

val rdd: RDD[(Int, String)] = sc.parallelize(Seq(
    (1, "aa"), 
    (2, "ab"), 
    (3, "ac") 
)) 

// use "cartesian", then "collect" to map only relevant results 
val result: RDD[Int] = rdd.cartesian(rdd).collect { 
    case ((t1: Int, _), (t2: Int, _)) if t1 != t2 => t1 * t2 
} 

注：この実装は、指示通りに入力レコードが、一意である前提としています。一致しない場合は、値の代わりにインデックスを比較しながら、デカルト結合とrdd.zipWithIndexの結果のマッピングを実行できます。