Hadoop的数据压缩

一、hadoop的数据压缩

　　1.概述

　　　　在进行mr程序的过程中，在mapper和reducer端会发生大量的数据传输和磁盘io，如果在这个过程中对数据进行压缩处理，可以有效的减少底层存储(hdfs)读写的字节数，，并且通过减少map和reduce阶段数据的输入输出来提升mr程序的速度，提高了网络带宽和磁盘空间的效率；

　　　　数据压缩可以有效的节省资源，它是mr程序的优化策略之一；　　

　　　　数据压缩会增加cpu的计算负担，但是能很大程度较少磁盘的io。由于数据压缩占用cpu资源很小，总体还是利大于弊的。

　　2.数据压缩使用原则：

　　　　运算密集型的任务尽量少用压缩、io密集型的任务多用压缩。

　　3.mapreduce支持的压缩编码

hadoop中的压缩格式	是否自带	文件的拓展名	是否可以切分
default	是	.default	否
gzip	是	.gz	否
bzip2	是	.bz2	是
lzo	否	.lzo	是
snappy	否	.snappy	否

　　4.编码解码器

default	org.apache.hadoop.io.compress.defaultcodec
gzip	org.apahce.hadoop.io.compress.gziocodec
bzop2	org.apache.hadoop.io.compress.bzio2codec
lzo	com.apache.hadoop.compression.lzocodec
snappy	org.apache.hadoop.io.compress.snappycodec

　　5.压缩性能

压缩格式	原始文件	压缩后文件	压缩速度	解压速度
gzip	8.3g	1.8g	17.5mb/s	58mb/s
bzip2	8.3g	1.1g	2.4mb/s	9.5mb/s
lzo	8.3g	2.9g	49mb/s	74.6mb/s

 

 

二 、hadoop压缩的使用

 

　　1.应用在wordcount程序中

　　　1）在map端对数据进行压缩

　　　　在driver类中的获取job对象后加入配置信息：

//开启map端的输入压缩
conf.setboolean("mapreduce.map.output.compress",true);

//设置压缩方法
//默认
conf.setclass("mapreduce.map.output.compress.codec",defaultcodec.class,compressioncodec.class);
//bzip2
conf.setclass("mapreduce.mapt.output.compress.codec",bzip2codec.class,compressioncode.class);
//lzo
conf.setclass("mapreduce.map.output.compress.codec".lzocodec.class,compressioncodec.class);

 

　　　　注意：在map端开启压缩并不能从结果文件中看到结果的改变，只要程序运行成功就代表设置没问题！

　　　2）在reduce端对数据进行压缩

　　　　在设置reduce输出数据类型之后加入配置信息：

//开启reduce端的输出压缩
fileoutputformat.setcompressoutput(job,true);
//设置压缩方法
//默认
fileoutputformat.setoutputcompressorclass(job,defaultcodec.class);
//bzip2
fileoutputformat.setoutputcompressorclass(job,bzip2codec.class);
//gzip
fileoutputformat.setoutputcompressorclass(job,gzipcodec.class);

　　　　三种选择一种即可，可以看到对应的结果文件看到被压缩的结果文件。

　　2.自定义压缩方法

 

/**
 * @author: princesshug
 * @date: 2019/4/8, 9:49
 * @blog: https://www.cnblogs.com/hellobigtable/
 */
public class testcompress {
    public static void main(string[] args) throws ioexception, classnotfoundexception {
        compress("g:\\weblog.log","org.apache.hadoop.io.compress.bzip2codec");
    }

    //自定义压缩方法
    private static void compress(string filename,string method) throws ioexception, classnotfoundexception {
        //获取输入流
        fileinputstream fis = new fileinputstream(new file(filename));

        //通过反射获取压缩方法并初始化
        class cname = class.forname(method);
        compressioncodec codec = (compressioncodec) reflectionutils.newinstance(cname, new configuration());

        //定义输出流
        fileoutputstream fos = new fileoutputstream(new file(filename + codec.getdefaultextension()));

        //创建压缩输出流
        compressionoutputstream cos = codec.createoutputstream(fos);

        //流的拷贝
        ioutils.copybytes(fis,cos,2*1024*1024,false);

        //关闭资源
        fis.close();
        cos.close();
        fos.close();
    }
}