大数据（十五）：Hadoop数据压缩与压缩/解压缩实例

一、数据压缩

1.概论

        压缩技术能够有效减少低层存储系统（HDFS）读写字节。压缩提高了网络带宽和磁盘空间的效率。在Hadoop下，尤其是数据规模很大和工作负载密集的情况下。使用数据压缩闲的非常重要。在这种情况下，I/O操作和网络数据传输要花大量的时间。还有，shuffle与merge过程同样也面临着巨大的I/O压力。

        磁盘I/O和网络带宽是Hadoop的宝贵资源，数据压缩对于节省资源、最小化磁盘I/O和网络传输是非常有帮助。不过，尽管压缩与解压操作的CPU开销不高，其性能的提升和资源节省也并非没有任何代价。
        如果磁盘I/O和网络带宽影响了MapReduce作业性能，在任意MapReduce阶段启用压缩都可以该生端到端处理时间并减少I/O和网络流量。

        压缩MapReduce的一种优化策略：通过压缩编码对Mapper或者Reducer的输出进行压缩，以减少磁盘I/O，提高MR程序运行速度，但同时也会增大cpu运算负担。

        压缩特性运用得当能提高新能，但运用不当也可能降低性能。

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

2. I/O密集型的job，多用压缩

2.MR支持的常见压缩编码

压缩格式	hadoop是否自带	算法	文件扩展名	是否可切分	换成压缩格式后，原来的程序是否需要修改
DEFAULT	是	DEFAULT	deflate	否	和文本处理一样，无需修改
Gzip	是	DEFAULT	gz	否	和文本处理一样，无需修改
bzip2	是	bzip2	bz2	是	和文本处理一样，无需修改
LZO	否	LZO	lzo	是	需要建立索引，还需要指定输入格式
Snappy	否	Snappy	snappy	否	和文本处理一样，无需修改

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

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

压缩性能比较

压缩算法	原生文件大小	压缩后文件大小	压缩速度	解压速度
gzip	8.3GB	1.8GB	17.5MB/s	58MB/s
bzip2	8.3GB	1.1GB	2.4MB/s	9.5MB/s
LZO	8.3GB	2.9GB	49.3MB/s	74.6MB/s
Snappy			250MB/s	500MB/s

3.Gzip压缩

        优点：压缩率比较高，而且压缩/解压速度也比较块：hadoop本身支持，在应用中处理gzip格式的文件就和直接处理文本一样；大部分linux系统都自带gzip命令，使用方便。

        缺点：不支持分区（split）

        应用场景：每当文件压缩之后在130m以内的，都可以考虑用Gzip压缩格式，hive程序，Streaming程序和java写的MapReduce程序完全和文本处理一样，压缩之后原来的程序不需要做任何修改。

4.Bzip2压缩

        优点：支持Split，具有很高的压缩率，比Gzip压缩率都高，hadoop本身支持，但不支持native，在linux系统下自带bzip2命令，使用方便。

        缺点：压缩、解压速度慢，不支持native

        应用场景：适合对速度要求不高，但需求较高的压缩率的时候，可以作为MapReduce作业的输出格式，或者输出之后的数据比较大，处理之后的数据需要压缩文档减少磁盘空间并且以后数据用的比较少的情况，或者对单个很大的文本文件想要压缩减少存储空间，同时又需要支持split，而且兼容之前的应用程序的情况。

5.Lzo压缩

        优点：压缩/解压速度也比较快，合理的压缩率，支持split，是hadoop中最流行的压缩格式，可以在linux系统下安装lzop命令，使用方便。

        缺点：压缩比例比gzip要低一些，hadoop本身并不支持，需要安装，在应用中对lzo格式的文件需要做一些特殊处理。（为了支持split需要建立索引，还需要指定inputformat为lzo格式）

        应用场景：一个很大的文本文件，压缩之后还大于200m以上的可以考虑，而且单个文件越大，lzo优点越明显。

6.Snappy压缩

        优点：高速压缩速度和合理的压缩率

        缺点：不支持Split，压缩率比gzip低，hadoop本身不支持

        应用场景：当MapReduce作业的Map输出的数据比较大的时候，作为Map到Reduce的中间数据的压缩格式，或者作为一个MapReduce作业的输出和另外一个MapReduce作业的输入

7.压缩配置参数

在Hadoop中启用压缩，可以配置以下参数（mapred-site.xml文件中）：

 

参数	默认值	阶段	建议
io.compression.ccodecs（在core-site.xml中配置）	org.apache.hadoop.io.compress.DefaultCodec, org.apache.hadoop.io.compress.GzipCodec, org.apache.hadoop.io.compress.BZip2Codec, org.apache.hadoop.io.compress.Lz4Codec	输入压缩	Hadoop使用文件扩展名判断是否支持某种编码、解码器
mapreduce.map.output.compress	false	mapper输出	这个参数设置为true启动压缩
mapreduce.map.output.compress.codec	org.apache.hadoop.io.compress.DefaultCodec	mapper输出	使用LZO或Snappy编码、解码器在这个阶段进行压缩
mapreduce.output.fileoutputformat.compress.codec	org.apache.hadoop.io.compress.DefaultCodec	reducer输出
mapreduce.output.fileoutputformat.compress.type	RECORD	reducer输出	SequenceFile输出使用的压缩类型：NONE和BLOCK

二、压缩、解压缩实例

        CompressionCodec有两个方法可以用于轻松地压缩或解压缩数据。要想对正在被写入一个输出流的数据进行压缩，我们可以使用createOutputStream（OutputStreamout）方法创建一个CompressionOutputStream,将其以压缩格式写入低层的流。相反，想要对输入流读取而来的数据进行解压缩，则需要调用createInputStream(InputStreamin)函数，从而获取一个CompressionInputStream，从而从低层的流读取未压缩的数据。
1.压缩实例

public class TestCompress {
    public static void main(String[] args) throws IOException, ClassNotFoundException {
        //测试压缩
        compress("d:/hello.txt","org.apache.hadoop.io.compress.Bzip2Codec");
    }

    private static void compress(String fileName, String method) throws IOException, ClassNotFoundException {
        //获取输入流
        FileInputStream fileInputStream = new FileInputStream(new File(fileName));
        //反射获取压缩类的信息
        Class<?> className = Class.forName(method);
        CompressionCodec codec = (CompressionCodec)ReflectionUtils.newInstance(className, new Configuration());
        //获取输出流
        FileOutputStream fileOutputStream = new FileOutputStream(new File(fileName + codec.getDefaultExtension()));
        CompressionOutputStream cos = codec.createOutputStream(fileOutputStream);
        //流的对拷
        IOUtils.copyBytes(fileInputStream,cos,1024*1024*5,false);
        //关闭流
        fileInputStream.close();
        cos.close();
        fileOutputStream.close();
    }
}

2.解压缩实例

public class TestDeCompress {
    public static void main(String[] args) throws IOException {
        decompress("d:/hello.txt.bz2");
    }

    private static void decompress(String fileName) throws IOException {
        //校验是否支持解压
        CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());
        CompressionCodec codec = factory.getCodec(new Path(fileName));
        if (codec == null){
            //不支持解码
            return;
        }
        //获取输入流
        CompressionInputStream cis = codec.createInputStream(new FileInputStream(new File(fileName)));
        //获取输出流
        FileOutputStream fos = new FileOutputStream(new File(fileName + ".decode"));

        //流对拷
        IOUtils.copyBytes(cis,fos,1024*1024*5,false);
        //关闭流
        cis.close();
        fos.close();
    }
}

3.Map端输出采用压缩

在Driver中获取job之前启动压缩并设置压缩方式

//开启map端的压缩
configuration.setBoolean("mapreduce.map.output.compress",true);
//设置map端输出压缩方式
configuration.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

4.在Reducer端输入采用压缩

//设置reduce端输出压缩开启
FileOutputFormat.setCompressOutput(job,true);
//设置压缩的方式
FileOutputFormat.setOutputCompressorClass(job,BZip2Codec.class);