MapReduce(十)：压缩

概述

1）压缩的好处和坏处

压缩的优点：以减少磁盘IO、减少磁盘储存空间

压缩的缺点：增加CPU开销

2）压缩的原则

（1）运算密集型的job，少用压缩

（2）IO密集型的job，多用压缩

MR支持的压缩编码

1）压缩算法对比介绍

压缩格式	Hadoop自带？	算法	文件扩展名	是否可切片	换成压缩格式后，<br />原来的程序是否需要修改
DEFLATE	是，直接使用	DEFLATE	.deflate	否	和文本处理一样，不需要修改
Gzip	是，直接使用	DEFLATE	.gz	否	和文本处理一样，不需要修改
bzip2	是，直接使用	bzip2	.bz2	是	和文本处理一样，不需要修改
LZO	否，需要安装	LZO	.lzo	是	需要建索引，还需要指定输入格式
Snappy	是，直接使用	Snappy	.snappy	否	和文本处理一样，不需要修改

压缩方式选择

压缩方式选择时重点考虑：压缩/解压缩速度、压缩率（压缩后存储大小）、压缩后是否可以支持切片。s

Gzip压缩

优点：压缩率比较高

缺点：不支持Split；压缩/解压缩速度一般

Bzip2压缩

优点：压缩率高；支持Split

缺点：压缩/解压缩速度慢

Lzo压缩

优点：压缩/解压缩速度比较块；支持Split

缺点：压缩率一般；想支持切片需要额外创建索引

Snappy压缩

优点：压缩和解压缩速度快

缺点：不支持Split，压缩率一般

MapReduce数据压缩

Map输入端采用压缩

无需显示指定使用的编解码方式。Hadoop自动检查文件扩展名，如果扩展名能够匹配，就会用恰当的编解码方式对文件进行压缩和解压。

企业开发：考虑因素

1）数据量小于块大小，重点考虑压缩和解压缩速度比较快的LZO/Snappy。

2）数据量非常大，重点考虑支持切片的Bzip2和LZO。

Map输出采用压缩

企业开发中如何选择：为了减少MapTask和ReduceTask之间的网络IO。重点考虑压缩和解压缩快的LZO、Snappy。

Reduce输出采用压缩

看需求：

• 如果数据永久保存，考虑压缩率比较高的Bzip2和Gzip。
• 如果作为下一个MapReduce输入，需要考虑数据量和是否支持切片。

压缩参数配置

1）为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器

压缩格式	对应的编码/解码器
DEFLATE	org.apache.hadoop.io.compress.DefaultCodec
gzip	org.apache.hadoop.io.compress.GzipCodec
bzip2	org.apache.hadoop.io.compress.BZip2Codec
LZO	com.hadoop.compression.lzo.LzopCodec
Snappy	org.apache.hadoop.io.compress.SnappyCodec

2）要在Hadoop中启用压缩，可以配置如下参数

参数	默认值	阶段	建议
io.compression.codecs （在core-site.xml中配置）	无，这个需要在命令行输入hadoop checknative查看	输入压缩	Hadoop使用文件扩展名判断是否支持某种编解码器
mapreduce.map.output.compress（在mapred-site.xml中配置）	false	mapper输出	这个参数设为true启用压缩
mapreduce.map.output.compress.codec（在mapred-site.xml中配置）	org.apache.hadoop.io.compress.DefaultCodec	mapper输出	企业多使用LZO或Snappy编解码器在此阶段压缩数据
mapreduce.output.fileoutputformat.compress（在mapred-site.xml中配置）	false	reducer输出	这个参数设为true启用压缩
mapreduce.output.fileoutputformat.compress.codec（在mapred-site.xml中配置）	org.apache.hadoop.io.compress.DefaultCodec	reducer输出	使用标准工具或者编解码器，如gzip和bzip2

实战

WordCountMapper.java

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    private Text outKey = new Text();
    private IntWritable outV = new IntWritable(1);

    @Override
    public void map(LongWritable key, Text value,
        Mapper<LongWritable, Text, Text, IntWritable>.Context context)
        throws IOException, InterruptedException {
        // 1 获取一行
        String lineStr = value.toString();
        // 2 切割
        String[] words = lineStr.split(" ");
        // 3 循环写出
        for (String word : words) {
            // 封装outKey
            outKey.set(word);
            // 写出
            context.write(outKey, outV);
        }
    }
}

WordCountReducer.java

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    IntWritable outV = new IntWritable();

    @Override
    public void reduce(Text key, Iterable<IntWritable> values,
        Reducer<Text, IntWritable, Text, IntWritable>.Context context)
        throws IOException, InterruptedException {
        int sum = 0;
        // 累加
        for (IntWritable value : values) {
            sum += value.get();
        }
        outV.set(sum);
        // 写出
        context.write(key, outV);
    }
}

WordCountDriver.java

public class WordCountDriver {

    public static void main(String[] args) throws Exception {

        //1 获取job
        Configuration conf = new Configuration();
        // 开启map端输出压缩
        conf.setBoolean("mapreduce.map.output.compress", true);

        // 设置map端输出压缩方式
        conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);
        Job job = Job.getInstance(conf);
        //2 设置jar包路径
        job.setJarByClass(WordCountDriver.class);
        //3 关联mapper、reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        //4 设置mapper输出的kv类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        //5 设置最终输出的kv类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        //6 设置输入路径和输出路径
        FileInputFormat.setInputPaths(job, new Path(System.getProperty("user.dir")+"/input/yasuo"));
        FileOutputFormat.setOutputPath(job, new Path(System.getProperty("user.dir")+"/output/yasuo"));
        // 设置reduce端输出压缩开启
        FileOutputFormat.setCompressOutput(job, true);

        // 设置压缩的方式
        // FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);
       // FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
       FileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class);
        //7 提交job
        Boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

小结：

通过本届学习，我们了解到Hadoop压缩的三个阶段，分别为map前、map后及reduce后；了解5中压缩算法，其中LZO和Snapppy最为常用。