Hadoop MapReduce中压缩技术的使用

Compression and Input Splits
 
当我们使用压缩数据作为MapReduce的输入时,需要确认数据的压缩格式是否支持切片?
 
假设HDFS中有一个未经压缩的大小为1GB的文本文件,如果HDFS Block大小为128MB,那么这个文件会被HDFS存储为8个Block。当MapReduce Job使用这个文件作为输入时将会创建8个切片(默认每一个Block生成一个切片),每一个切片关联的数据都可以被一个Map Task独立地处理。
 
如果这个文本文件使用Gzip格式压缩,大小仍为1GB,如前所述,它也会被HDFS存储为8个Block。可是当MapReduce Job使用这个文件作为输入时,为每一个Block生成一个切片是不可行的,取而代之的是整个文件将作为一个切片被一个Map Task所处理。
 
Gzip使用DEFLATE存储压缩数据,DEFLATE将数据存储为一系列的压缩数据块,可是这些压缩数据块的边界是无法区分的,导致在数据流中无法定位某个数据块的起始位置。也就是说,我们无法随意地指定一个位置(该位置不一定恰好是某数据块的起始位置),然后移动到下一个数据块的起始位置读取数据。基本这个原因,Gzip格式的文件是不支持切片的。
 
对于这种情况,MapReduce是可以作出正确处理的,通过文件后缀名(文件后缀名直接影响压缩数据格式的判断)可以判断出这个文件是以Gzip格式进行压缩的,不支持切片,会将整个文件作为一个切片进行处理。但这样做是有很大代价的,一个Map Task要处理整个文件的数据,而且大部分数据并不是“数据本地性”的。
 
如果这个文本文件使用LZO格式压缩,同样的问题也会存在,但是Hadoop LZO Library提供了一个用于预处理LZO文件的切片索引工具,可以简单地认为生成的索引文件中保存着各个切片的起始位置,再配合合适的InputFormat(如:LzoTextInputFormat),运行MapReduce Job时就可以支持切片。
 
Bzip2的各个数据块之间存放有专门的“Synchronization Marker”,因此它是可以支持切片的。
 
Hadoop通常处理的都是大规模的数据集,因此我们必须尽可能的利用压缩优化性能。具体使用哪一个压缩格式依赖于文件大小、文件格式以及我们使用的分析工具。以下是一些使用建议:
 
(1)使用一些容器文件格式,如Sequence File、Avro DataFile、ORCFile、Parquet File,这些文件格式全部支持压缩和切片,配合一个快速的压缩算法(如:LZO、LZ4、Snappy)使用通常是一个好的选择;
 
(2)使用一个支持切片的压缩算法,如bzip2、lzo(通过索引处理之后可以支持切片);
 
(3)将一个文件人为地切分为Chunk(即一个文件被切分为多个文件),然后将这些Chunks逐个的进行压缩,可以使用任意支持的压缩算法,且不需要考虑压缩算法是否支持切片,但是Chunk压缩后的大小需要接近于HDFS Block的大小;
 
(4)文件不作压缩处理。
 
对于大型的文件,我们不能选择不支持切片的压缩算法,因为这会导致MapReduce Job丧失数据本地性且运行效率低下。
 
Using Compression in MapReduce
 
MapReduce读取输入路径中的压缩文件时会自动完成数据解压(可参考CompressionCodecFactory)。
 
如果MapReduce Job的结果输出需要使用压缩,可以通过设置Job的相关配置属性来实现:
 
mapreduce.output.fileoutputformat.compress:true
 
mapreduce.output.fileoutputformat.compress.codec:CompressionCodec全限定类名
 
也可以通过FileOutputFormat提供的静态方法设置,如:
 
FileOutputFormat.setCompressOutput(job, true);
FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
 
不同的输出文件格式可能相应的设置属性会有不同。
 
Compressing map output
 
Map Task的输出被写出到本地磁盘,而且需要通过网络传输至Reduce Task的节点,只要简单地使用一个快速的压缩算法(如LZO、LZ4、Snappy)就可以带来性能的提升,因为压缩机制的使用避免了Map Tasks与Reduce Tasks之间大量中间结果数据被传输。可以通过设置相应的Job配置属性开启:
 
mapreduce.map.output.compress:true
 
mapreduce.map.output.compress.codec:CompressionCodec全限定类名
 
也可以通过Configuration API进行设置:
 
new API:
 
Configuration conf = new Configuration();
conf.setBoolean(Job.MAP_OUTPUT_COMPRESS, true);
conf.setClass(Job.MAP_OUTPUT_COMPRESS_CODEC, GzipCodec.class, CompressionCodec.class);
Job job = new Job(conf);
 
old API:
 
conf.setCompressMapOutput(true);
conf.setMapOutputCompressorClass(GzipCodec.class);
 
 
 
 
 
 
 
 
上一篇:Cookies and Caching Client Identification


下一篇:bzoj 1735: [Usaco2005 jan]Muddy Fields 泥泞的牧场 最小点覆盖