クラスタ上でファイルを保存する方法

Question

私はsshを使用してクラスタに接続されていると私は私がテキストファイルに結果を保存したい

spark-submit --master yarn myProgram.py 

使用してクラスタにプログラムを送信し、私が使用してみました次の行：

counts.write.json("hdfs://home/myDir/text_file.txt") 
counts.write.csv("hdfs://home/myDir/text_file.csv") 

ただし、いずれも機能しません。プログラムが終了し、テキストファイルがmyDirに見つかりません。どのように私はこれを行うことができますか考えていますか？

また、ローカルマシンに直接書き込む方法はありますか？

編集：私はhomeディレクトリは今、私はその結果を保存するように存在していないことが判明： counts.write.json("hdfs:///user/username/text_file.txt") しかし、これはtext_file.txtという名前のディレクトリを作成し、内部の私は内部の部分的な結果を持つファイルがたくさんあります。しかし、最終的な結果を含む1つのファイルが必要です。どのように私はこれを行うことができます任意のアイデア？

Answer 1

スパークは、計算が分散されているので結果を複数のファイルに保存します。したがって、書き込み：

counts.write.csv("hdfs://home/myDir/text_file.csv") 

はフォルダtext_file.csvで別のファイルとして各パーティション上のデータを保存することを意味します。あなたは、単一のファイルとして保存されたデータが必要な場合は、最初のcoalesce(1)を使用します。

counts.coalesce(1).write.csv("hdfs://home/myDir/text_file.csv") 

これは、単一のパーティションにすべてのデータを置くと、保存されたファイルの数はこれが1になります。これは悪いかもしれませんあなたが多くのデータを持っているなら、アイデア。データが非常に小さい場合は、collect()を使用してください。これにより、すべてのデータがドライバマシンに配列として格納され、その後、単一のファイルとして保存されます。

Answer 2

エラーが発生しますか？おそらく、あなたがそのフォルダからの書き込み/読み取りのための正しいアクセス権を持っているかどうかを確認することができます。

また、既定でSparkは、あなたが持っているパーティションの数に応じて、いくつかのファイルを含むtext_file.txtという名前のフォルダを作成します。

ローカルマシンに書き込む場合は、パスをfile:///home/myDir/text_file.txtで指定できます。 /user/hdfs/...のようなパスを使用すると、デフォルトでHDFSに書き込まれます

Answer 3

固有のファイル（必要に応じて指定されていない）を使用するには、.repartition(1)、look hereをRDDに転送する必要があります。 あなたのhdfsパスが間違っていると思います。 Sparkではテキストファイル用のHDFSがデフォルトであり、Hadoop（デフォルトでは）は以前作成していない限り、root dirにホームディレクトリはありません。 csv/txtファイル（この拡張機能付き）を書き込む唯一の方法は、RDDやDF関数を使用しないで、通常のPython csvとioのライブラリを使用している場合は、でRDDをmartix（データセットは巨大ではありません）。

あなたが（とは、HDFS上の）ファイルシステムに直接書き込む場合は

counts.write.csv("file:///home/myDir/text_file.csv") 

を使用しかし、これはcsvファイル拡張子を持つ1つのファイルを書き込みません。それは、あなたのデータセットのn個のパーティションのうちの一部m-0000nを持つフォルダを作成します。

Answer 4

あなたは、コマンドラインから一つのファイルにあなたの結果を連結することができます

hadoop fs -cat hdfs:///user/username/text_file.txt/* > path/to/local/file.txt 

これは​​3210を使用するよりも速くなければなりません - すべてのデータが通過注ぎ込まれているので、私の経験では、すべてのcollect()タイプの操作が遅いですマスターノード。さらに、データがマスタノードのメモリを超えている場合は、collect()で問題に遭遇する可能性があります。

しかし、の場合、この方法では、以前の実行からファイルを明示的に削除する必要があります（現行の実行では正確に同じ数のファイルが生成されない可能性があります）。それぞれの実行でこれを行うフラグがあるかもしれませんが、わかりません。

削除するには：

hadoop fs -rm -r hdfs:///user/username/text_file.txt/*