Hadoop的數據壓縮

一、Hadoop的數據壓縮

　　1.概述

　　　　在進行MR程式的過程中，在Mapper和Reducer端會發生大量的數據傳輸和磁碟IO，如果在這個過程中對數據進行壓縮處理，可以有效的減少底層存儲(HDFS)讀寫的位元組數，，並且通過減少Map和Reduce階段數據的輸入輸出來提升MR程式的速度，提高了網路帶寬和磁碟空間的效率；

　　　　數據壓縮可以有效的節省資源，它是MR程式的優化策略之一；　　

　　　　數據壓縮會增加cpu的計算負擔，但是能很大程度較少磁碟的IO。由於數據壓縮占用cpu資源很小，總體還是利大於弊的。

　　2.數據壓縮使用原則：

　　　　運算密集型的任務儘量少用壓縮、IO密集型的任務多用壓縮。

　　3.MapReduce支持的壓縮編碼

　　4.編碼解碼器

　　5.壓縮性能

二 、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();
    }
}