压缩Spark中的序列文件？

Question

我正在尝试将RDD保存为压缩的Sequence文件.我可以保存非压缩文件来调用:

counts.saveAsSequenceFile(output)

其中count是我的RDD(IntWritable,Text).但是,我没有设法压缩输出.我尝试了几种配置并且总是遇到异常:

counts.saveAsSequenceFile(output, classOf[org.apache.hadoop.io.compress.SnappyCodec])
<console>:21: error: type mismatch;
 found   : Class[org.apache.hadoop.io.compress.SnappyCodec](classOf[org.apache.hadoop.io.compress.SnappyCodec])
 required: Option[Class[_ <: org.apache.hadoop.io.compress.CompressionCodec]]
              counts.saveAsSequenceFile(output, classOf[org.apache.hadoop.io.compress.SnappyCodec])

 counts.saveAsSequenceFile(output, classOf[org.apache.spark.io.SnappyCompressionCodec])
<console>:21: error: type mismatch;
 found   : Class[org.apache.spark.io.SnappyCompressionCodec](classOf[org.apache.spark.io.SnappyCompressionCodec])
 required: Option[Class[_ <: org.apache.hadoop.io.compress.CompressionCodec]]
              counts.saveAsSequenceFile(output, classOf[org.apache.spark.io.SnappyCompressionCodec])

它甚至对Gzip也不起作用:

 counts.saveAsSequenceFile(output, classOf[org.apache.hadoop.io.compress.GzipCodec])
<console>:21: error: type mismatch;
 found   : Class[org.apache.hadoop.io.compress.GzipCodec](classOf[org.apache.hadoop.io.compress.GzipCodec])
 required: Option[Class[_ <: org.apache.hadoop.io.compress.CompressionCodec]]
              counts.saveAsSequenceFile(output, classOf[org.apache.hadoop.io.compress.GzipCodec])

你能建议解决方案吗？另外,我没有发现如何指定压缩参数(即Snappy的压缩类型)

Answer 1

签名saveAsSequenceFile是def saveAsSequenceFile(path: String, codec: Option[Class[_ <: CompressionCodec]] = None).你需要发一个Option[Class[_ <: CompressionCodec]]as codec.例如,

counts.saveAsSequenceFile(output, Some(classOf[org.apache.hadoop.io.compress.SnappyCodec]))

如果您type mismatch仔细阅读了错误信息,您应该自己更正.