Spark Standalone Mode: как сжать искровый выход, записанный на HDFS

Связано с моим другим вопросом, но отличается:

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

Если я сохраню RDD в HDFS, как я могу сказать искру, чтобы сжать выход с помощью gzip? В Hadoop можно установить

mapred.output.compress = true

и выберите алгоритм сжатия с

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

Как бы я сделал это в искру? Будет ли это работать?

изменить: использование spark-0.7.2

Ответ 1

В методе saveAsTextFile используется дополнительный необязательный параметр класса codec. Поэтому для вашего примера это должно быть что-то вроде этого, чтобы использовать gzip:

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

UPDATE

Поскольку вы используете 0.7.2, вы можете перенести код сжатия через параметры конфигурации, которые вы установили при запуске. Я не уверен, что это будет работать точно, но вам нужно перейти от этого:

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)

примерно так:

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

Если вы его заработаете, размещение вашей конфигурации, вероятно, будет полезно и для других.

Ответ 2

Еще один способ сохранить gzip файлы в HDFS или систему каталогов Amazon S3 - использовать метод saveAsHadoopFile.

someMap - это RDD [(K, V)], если у вас есть someMap как RDD [V], вы можете вызвать someMap.map(line = > (line, ""), чтобы использовать метод saveAsHadoopFile.

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

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

Ответ 3

Для более новой версии Spark выполните следующие действия в файле spark-defaults.xml. (mapred прекращается).

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