所有与文本相关的数据源,包括CSVDataSource,都使用Hadoop文件API来处理文件(这也是Spark Core的RDD中使用的方法)。
您可以在readFile中找到相关行,该行导致HadoopFileLinesReader,其包含以下行:
val fileSplit = new FileSplit(
new Path(new URI(file.filePath)),
file.start,
file.length,
Array.empty)
这使用了Hadoop的org.apache.hadoop.fs.Path,处理底层文件的压缩。
经过快速搜索,我找到了处理压缩的Hadoop属性,即mapreduce.output.fileoutputformat.compress
。
这引导我查找Spark SQL的CompressionCodecs,该配置具有以下压缩配置:
"none" -> null,
"uncompressed" -> null,
"bzip2" -> classOf[BZip2Codec].getName,
"deflate" -> classOf[DeflateCodec].getName,
"gzip" -> classOf[GzipCodec].getName,
"lz4" -> classOf[Lz4Codec].getName,
"snappy" -> classOf[SnappyCodec].getName)
在下面的代码中,您可以找到
setCodecConfiguration,它使用了“我们”的选项。
def setCodecConfiguration(conf: Configuration, codec: String): Unit = {
if (codec != null) {
conf.set("mapreduce.output.fileoutputformat.compress", "true")
conf.set("mapreduce.output.fileoutputformat.compress.type", CompressionType.BLOCK.toString)
conf.set("mapreduce.output.fileoutputformat.compress.codec", codec)
conf.set("mapreduce.map.output.compress", "true")
conf.set("mapreduce.map.output.compress.codec", codec)
} else {
conf.set("mapreduce.output.fileoutputformat.compress", "false")
conf.set("mapreduce.map.output.compress", "false")
}
}
另一种方法
getCodecClassName用于解析
JSON、
CSV和
text格式的
compression
选项。