Spark trybie autonomicznym: Jak kompresować wyjście zapłonową pisemnej do HDFS

Question

związane z moją inne pytanie, ale wyraźny:

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

Gdybym zapisać RDD do HDFS, jak mogę powiedzieć iskrę do kompresji wyjście z gzip ? W Hadoop, możliwe jest ustawienie

mapred.output.compress = true 

i wybrać algorytm kompresji z

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

Jak zrobić to w iskry? Czy to też zadziała?

edit: using iskra-0.7.2

Answer 1

Sposób saveAsTextFile trwa dodatkowy opcjonalny parametr klasy kodeka aby wykorzystać. Tak dla przykładu powinno być coś takiego w użyciu gzip:

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

UPDATE

Ponieważ używasz 0.7.2 może być w stanie do portu kod kompresji poprzez opcje konfiguracyjne, które ustawić przy starcie. Nie jestem pewien, czy to będzie działać dokładnie, ale trzeba iść z tego:

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) 

do czegoś takiego:

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") 

Jeśli zmusić go do pracy, publikując swój config prawdopodobnie być także pomocnym dla innych.

Answer 2

Innym sposobem na zapisanie skompresowanych plików gzip do systemu plików HDFS lub Amazon S3 jest użycie metody saveAsHadoopFile.

someMap jest RDD [(K v)], jeśli someMap jak RDD [V] można nazwać someMap.map (wiersz => (linia "") do zastosowania metody 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

Dla nowszej wersji Spark, należy wykonać następujące czynności w pliku o zapłonie defaults.xml. (mapred jest 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>