Spark modalità Standalone: ​​Come comprimere uscita scintilla scritto HDFS

Question

correlati alla mia altra domanda, ma distinti:

someMap.saveAsTextFile("hdfs://HOST:PORT/out") 

Se salvo l'RDD per HDFS, come posso dire scintilla per comprimere l'output con gzip ? In Hadoop, è possibile impostare

mapred.output.compress = true 

e scegliere l'algoritmo di compressione con

mapred.output.compression.codec = <<classname of compression codec>> 

Come farei questo a scintilla? Funzionerà anche questo?

modificare: usando scintilla 0.7.2

Answer 1

Il metodo saveAsTextFile prende un parametro opzionale aggiuntiva della classe codec da utilizzare. Così, per il tuo esempio dovrebbe essere qualcosa di simile da usare gzip:

someMap.saveAsTextFile("hdfs://HOST:PORT/out", classOf[GzipCodec]) 

UPDATE

Dal momento che si sta utilizzando 0.7.2 si potrebbe essere in grado di porta il codice di compressione tramite le opzioni di configurazione che si impostato all'avvio. Non sono sicuro se questo funzionerà esattamente, ma è necessario andare da questo:

conf.setCompressMapOutput(true) 
conf.set("mapred.output.compress", "true") 
conf.setMapOutputCompressorClass(c) 
conf.set("mapred.output.compression.codec", c.getCanonicalName) 
conf.set("mapred.output.compression.type", CompressionType.BLOCK.toString) 

a qualcosa di simile:

System.setProperty("spark.hadoop.mapred.output.compress", "true") 
System.setProperty("spark.hadoop.mapred.output.compression.codec", "true") 
System.setProperty("spark.hadoop.mapred.output.compression.codec", "org.apache.hadoop.io.compress.GzipCodec") 
System.setProperty("spark.hadoop.mapred.output.compression.type", "BLOCK") 

Se si arriva a farlo funzionare, pubblicare la tua config probabilmente essere utile anche agli altri.

Answer 2

Un altro modo per salvare i file gzipped su HDFS o sul sistema di directory Amazon S3 consiste nell'utilizzare il metodo saveAsHadoopFile.

someMap è RDD [(K, V)], se avete someMap come RDD [V], è possibile chiamare someMap.map (line => (linea, "") di utilizzare il metodo saveAsHadoopFile.

import org.apache.hadoop.io.compress.GzipCodec 

someMap.saveAsHadoopFile(output_folder_path, classOf[String], classOf[String], classOf[MultipleTextOutputFormat[String, String]], classOf[GzipCodec]) 

Answer 3

Per diffusione Spark più recente, si prega di effettuare le seguenti operazioni nel file scintilla defaults.xml. (mapred è derecated).

<property> 
    <name>mapreduce.output.fileoutputformat.compress</name> 
    <value>true</value> 
</property> 
<property> 
    <name>mapreduce.output.fileoutputformat.compress.codec</name> 
    <value>GzipCodec</value> 
</property> 
<property> 
    <name>mapreduce.output.fileoutputformat.compress.type</name> 
    <value>BLOCK</value> 
</property>