各行に名前を含むファイルがあり、各行に順番に番号を追加したい。例えばのために ファイルは、私はそれがこのスパーク・クラスタファイル内のすべての行に順番に番号を付ける
a,1
b,2
c,3
d,4
ご存知のように、私はこの
val lines = sc.textFile("data.txt")
val pair = lines.zipWithIndex().map{case(i,line) => i.toString +","+line}
pair.collect()
を達成するために、このコードを記述しましたが、スパークを実現したい、この
a
b
c
d
のようなものである場合異なるクラスタ間でタスクを分散します。だから私はこれがうまくいくかどうかはわかりません。どうすればいいですか？前もって感謝します。
このコードを実行すると、期待している出力が得られます。 sparkがクラスタ全体にタスクを配布しても、プログラマチックには何も影響しません。例の場合、2ワーカーノードで実行している場合、ファイルは2つのパーティションに分割され、それぞれのワーカーノードに格納されます。今度はプログラムが実行され、ドライバがzipWithIndexを参照すると、zipWithIndexが必要なので、両方のワーカーにファイルの他のパーティションに関する情報が確実に渡されます。
異なる変換とアクションでは異なる要件があり、マスターノードでは、1つのコピーのみが存在することを確認するために必要なデータがシャッフルされていることを確認します。
もう1つのことは、行番号で単語のペアを作成したいだけで、マップは必要ありません。これだけでも動作します
上記のコード行をマップなしでjavaで実行したところ、正しい出力が得られました。行番号は0.で始まっていますが、作業ノードの数が行番号を順番に表示するのに何の影響も及ぼさないという点がまだ証明されています。
// output of worker 1 part-00000
a,0
b,1
//output of worker 2 part-00001
c,2
d,3
説明をありがとう – gashu
「」最初の行で、「B」は2行目にあるファイルdata.txtを内のデータと行番号 –
の関係は何です。だから基本的に私は行番号を追加しています – gashu
'zipWithIndex'に何か問題がありますか？タスクがノード間で分散されている場合でも期待通りに動作するはずです。 – Mariusz