2つの大規模RDDから新しいRDD変数を生成

Question

以下のように2つの大きなサイズのRDDがあります。

#First RDD 
key1 value1 labelA 
key2 value2 lableB 
..... 

#Second RDD 
stepA key1 value1 
stepB key2 value2 
... 

そして、私が抽出したいものは以下の通りです。

labelA stepA key1 value1 
labelB stepB key2 value2 

しかし、私の問題は2つのRDDサイズが非常に大きいことです。したがって、参加には多くの時間がかかることがあります。私は結合メソッドを避け、シャッフルサイズをできるだけ小さくしたいと考えています。また、1つのRDDを収集し、それをブロードキャスト変数にすることは、そのサイズのために機能しませんでした。 RDDサイズが10ギガバイトを超えていることに注意してください。これは、ドライバがそれらを一度に格納できないことを意味します。

したがって、2つの大きなサイズのRDDから新しいRDDを生成する方法はありますか？上記のように、私は結合メソッドを避けたい。

ありがとうございました。

Answer 1

3タプルのRDDがあると仮定すると、これで必要なものが得られます。

val left_rdd = rdd1 
    .map{ case (key, value, label) => (key, value) -> label } 
val right_rdd = rdd2 
    .map{ case (step, key, value) => (key, value) -> step } 
left_rdd.join(right_rdd) 
    .map{ case ((key, value), (label, step)) => (label, step, key, value) } 

長すぎると想定する前に試してください。 10GBはそれほど大きくありません。可能であれば、シャッフルは避けてくださいスパークしてください。しかし時には、それはちょうどあなたが必要とするものです。