十七、hadoop壓縮相關

一、數據壓縮在hadoop中的意義

1、基本概述

? 壓縮技術可以減少底層hdfs的讀寫字節數。并且能夠降低在數據傳輸過程中占用的網絡帶寬資源，以及降低占用的磁盤空間。而在MapReduce中，shuffle以及merge過程都面臨著巨大的IO壓力。但是要注意增加了壓縮，另外一方面會增加cpu的負載。所以在要權衡好是否采用壓縮，以及采用的壓縮算法的特性。

2、壓縮應用基本原則

運算密集型的job，少用壓縮。因為壓縮占用cpu。
IO密集型的job，可用壓縮減少數據量。
選擇壓縮算法時，要注意壓縮比，壓縮比越大，壓縮以及解壓時間越長。

二、MR支持的壓縮編碼格式

1、壓縮編碼

2、不同壓縮算法對應的編解碼器

3、不同壓縮算法的特性以及適用場景

（1）gzip

優點：
壓縮率高，解壓、壓縮速度也必比較快。hadoop本身自帶，在應用中處理gzip格式的文件就和直接處理文本一樣。大部分Linux自帶gzip命令，使用方便。

缺點：不支持split

適用場景：
當每個文件壓縮之后在一個block左右的大小（因為無法分片），都可以考慮使用gzip將原數據壓縮。例如可以將一天或者一小時的日志壓縮成一個gzip文件，運行MapReduce的時候就可以并行處理多個gzip。hive，streaming，MapReduce程序處理壓縮文件時，無需修改程序，就像處理文本文件一樣。

（2）bzip2

優點：
支持split；高壓縮比，比gzip高。hadoop自帶，Linux下自帶bzip2命令

缺點：壓縮、解壓速度慢，不支持native（java和C交互的api接口）

適用場景：
適合對速度要求不高，但需要較高的壓縮率的時候，可以作為mapreduce作業的輸出格式；或者輸出之后的數據比較大，處理之后的數據需要壓縮存檔減少磁盤空間并且以后數據用得比較少的情況；或者對單個很大的文本文件想壓縮減少存儲空間，同時又需要支持split，而且兼容之前的應用程序（即應用程序不需要修改）的情況。

（3）lzo

優點：
壓縮/解壓速度比較快，壓縮率合理（比gzip和bzip2小）。支持split，是hadoop中最流行的壓縮格式。可以在Linux下可以通過安裝lzop命令來使用

缺點：
壓縮率比gzip要低一些；hadoop本身不支持，需要安裝；在應用中對lzo格式的文件需要做一些特殊處理（為了支持split需要建索引，還需要指定inputformat為lzo格式）。

適用場景：
一個很大的文本文件，壓縮之后還大于200M以上的可以考慮，而且單個文件越大，lzo優點越越明顯。

（4）snappy

優點：壓縮和解壓速度快，合理的壓縮率
缺點：不支持的split，壓縮率比gzip要低；hadoop本身不支持，需要安裝
適用場景：
當Mapreduce作業的Map輸出的數據比較大的時候，作為Map到Reduce的中間數據的壓縮格式；或者作為一個Mapreduce作業的輸出和另外一個Mapreduce作業的輸入。

三、壓縮的配置

1、適用范圍

可以用在MapReduce的任意階段輸出、原始數據的壓縮、reduce的輸出等

2、hadoop壓縮配置參數

四、壓縮應用實戰例子

1、數據流的壓縮和解壓

package JavaCompress;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;

import java.io.*;

public class TestCompress {
    public static void main(String[] args) throws IOException, ClassNotFoundException {
        //compress("G:\\Fly Away-梁靜茹.mp3", "org.apache.hadoop.io.compress.GzipCodec");
        deCompress("G:\\Fly Away-梁靜茹.mp3.gz", "mp3");

    }

//壓縮
    public static void compress(String filename, String method) throws IOException, ClassNotFoundException {
        //創建輸入流
        FileInputStream fis = new FileInputStream(new File(filename));

        //通過反射獲取壓縮類的Class 對象
        Class codecClass = Class.forName(method);

        //通過反射獲取壓縮對象
        CompressionCodec codec = (CompressionCodec)ReflectionUtils.newInstance(codecClass, new Configuration());

        //創建普通輸出流對象
        FileOutputStream fos = new FileOutputStream(new File(filename + codec.getDefaultExtension()));

        //通過壓縮對象創建壓縮輸出流，類似于將普通輸出流封裝成壓縮輸出流
        CompressionOutputStream cos = codec.createOutputStream(fos);

        //流拷貝
        IOUtils.copyBytes(fis, cos, 1024 * 1024 * 5, false);

        fis.close();
        cos.close();
        fos.close();

    }

//解壓
    public static void deCompress(String filename, String decode) throws IOException {
        CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());

        //獲取文件的壓縮算法類型對象，返回值可用于檢查文件是否可解壓
        CompressionCodec codec = factory.getCodec(new Path(filename));
        if (codec == null) {
            System.out.println("不支持解壓縮：" + filename);
            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();

    }
}

2、map輸出采用壓縮

用法很簡單，只需在driver中給job配置以下參數即可

Configuration configuration = new Configuration();

// 開啟map端輸出壓縮
configuration.setBoolean("mapreduce.map.output.compress", true);
// 設置map端輸出壓縮方式
configuration.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

3、reduce輸出采用壓縮

依舊是在driver中設置以下

// 設置reduce端輸出壓縮開啟
FileOutputFormat.setCompressOutput(job, true);

// 設置壓縮的方式
FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class); 
//      FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); 
//      FileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class);