Spark/ScalaのHDFSへの書き込み

Question

私は、ZIPファイルを読み込み、解凍して内容を新しいファイルのセットに書き込むためのspark/scalaプログラムを作成しています。これをローカルファイルシステムに書き込むことができますが、HDFSなどの分散ファイルシステムに出力ファイルを書き込む方法があるのか​​どうか疑問に思っています。コードは（あなたが依存manangementツールとしてSBTを使用している場合、依存関係にthathライブラリを追加し、）あなたが簡単にHadoop-共通ライブラリを使用してHDFSにデータを書き込むことができ

import java.util.zip.ZipInputStream 
import org.apache.spark.input.PortableDataStream 
import java.io._ 

var i =1 
sc.binaryFiles("file:///d/tmp/zips/").flatMap((file: (String,  PortableDataStream)) => 
    { 


    val zipStream = new ZipInputStream(file._2.open)    
    val entry = zipStream.getNextEntry        
    val iter = scala.io.Source.fromInputStream(zipStream).getLines   

    val fname = f"/d/tmp/myfile$i.txt" 


    i = i + 1 

    val xx = iter.mkString 
    val writer = new PrintWriter(new File(fname)) 
    writer.write(xx) 
    writer.close() 

    iter              
    }).collect() 

`

Answer 1

below`示されています。そして、あなたは、例えば、Aのために、書くことができ

private val fs = { 
    val conf = new Configuration() 
    FileSystem.get(conf) 
    } 

は（などコア-site.xmlに、）あなたのHadoopクラスタ情報とファイルシステムを設定してください

：それを使用すると、FileSystemのオブジェクトを作成することができますパスの文字列（あなたのケースでは、あなたがストリームに対処する必要があります）、HDFS上で次のように：

@throws[IOException] 
    def writeAsString(hdfsPath: String, content: String) { 
    val path: Path = new Path(hdfsPath) 
    if (fs.exists(path)) { 
     fs.delete(path, true) 
    } 
    val dataOutputStream: FSDataOutputStream = fs.create(path) 
    val bw: BufferedWriter = new BufferedWriter(new OutputStreamWriter(dataOutputStream, "UTF-8")) 
    bw.write(content) 
    bw.close 
    } 

Answer 2

あなたは公式ドキュメントからの方法saveAsTextFileを見ている必要があります。http://spark.apache.org/docs/latest/programming-guide.html

それはあなたがHDFSに保存することができます：

iter.saveAsTextFile("hdfs://...") 

Answer 3

あなたはsaveAsTextFile方法を試すことができます。

データファイルの要素を、ローカルファイルシステム、HDFSまたは他のHadoopでサポートされているファイルシステムの特定のディレクトリにテキストファイル（またはテキストファイルのセット）として書き込みます。 Sparkは各要素のtoStringを呼び出して、ファイル内のテキスト行に変換します。

パーティションを別のファイルとして保存します。パーティションを分割したり、合体したりしない限り、最終的なパーティションの数は入力ファイルの数と同じになります。

Answer 4

sc.binaryFiles("/user/example/zip_dir", 10)        //make an RDD from *.zip files in HDFS 
      .flatMap((file: (String, PortableDataStream)) => {     //flatmap to unzip each file 
       val zipStream = new ZipInputStream(file._2.open)    //open a java.util.zip.ZipInputStream 
       val entry = zipStream.getNextEntry        //get the first entry in the stream 
       val iter = Source.fromInputStream(zipStream).getLines   //place entry lines into an iterator 
       iter.next              //pop off the iterator's first line 
       iter               //return the iterator 
      }) 
      .saveAsTextFile("/user/example/quoteTable_csv/result.csv")