背景

好久沒有更新了，原因是公司專案上線，差點被祭天。在這種驚心動魄的時候還是要抽時間做一點自己喜歡做的事情的，然而進度比預期慢了許多。

正式開始

接下來就開始記錄最近的學習成果啦！
在Hadoop叢集中，網路資源是非常珍貴的。因此對檔案進行壓縮是非常必要的，除此之外。壓縮檔案的另一個好處就是可以節省磁碟空間。

壓縮方法介紹

我們在對檔案進行壓縮時，要仔細考慮使用哪種壓縮方法最合適。常見的壓縮格式有

表5-1

相比較之下，bzip2比gzip更高效，但壓縮速度要慢一些。LZO壓縮速度比gzip更快，但效率不高。
在Hadoop中，一個對CompressionCodec介面的實現代表一個codec。例如，GzipCodec包裝了gzip的壓縮和解壓縮方法。（實際上每個Codec都包含了這兩個方法）
我們可以利用CompressionCodec中的方法對資料進行壓縮和解壓。
上表中“是否可切分”一列，表示壓縮演算法是否支援切分。可切分的壓縮格式更加適合MapReduce。gzip格式使用DEFLATE來儲存壓縮過的資料，DEFLATE將資料作為一系列壓縮過的塊進行儲存。問題是，每塊的開始沒有指定使用者在資料流中任意點定位到下一個塊的起始位置，而是其自身與資料流同步。因此，gzip並不支援檔案切分。但是，bzip2檔案提供不同資料塊之間的同步標識，因而它是支援切分的。

在MapReduce中使用壓縮

主要程式碼

這裡舉一個查詢每年最高氣溫的例子，對輸出資料進行壓縮。
Map

public class MaxTemperatureMapper extends MapReduceBase
    implements Mapper<LongWritable, Text, Text, IntWritable>{

    private static final int MISSING = 9999;

    public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
        String data = value.toString();
        String year = ""
 
;
        int airTemperature = 0;
        if(data != null && data.length() > 0) {
            String[] temp = data.split(",");
            year = temp[0];
            airTemperature = Integer.parseInt(temp[1]);
        }
        output.collect(new Text(year),new IntWritable(airTemperature));
    }
}

Reduce

public class MaxTemperatureReducer extends MapReduceBase
    implements Reducer<Text, IntWritable, Text, IntWritable>{

    public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
        int maxValue = Integer.MIN_VALUE;
        while (values.hasNext()) {
            maxValue = Math.max(maxValue, values.next().get());
        }
        output.collect(key, new IntWritable((maxValue)));
    }
}

Compression

public class MaxTemperatureWithCompression {
    public static void main(String[] args) throws IOException {
        if (args.length != 2) {
            System.err.println("Usage: MaxTemperatureWithCompression <input path> " +
                    "<output path>");
            System.exit(-1);
        }
        JobConf conf = new JobConf(MaxTemperatureWithCompression.class);
        conf.setJobName("Max temperature with output compression");

        FileInputFormat.addInputPath(conf, new Path(args[0]));
        FileOutputFormat.setOutputPath(conf, new Path(args[1]));

        conf.setOutputKeyClass(Text.class);
        conf.setOutputValueClass(IntWritable.class);

        conf.setBoolean("mapred.output.compress", true);
        conf.setClass("mapred.output.compression.codec", GzipCodec.class, CompressionCodec.class);

        conf.setMapperClass(MaxTemperatureMapper.class);
        conf.setCombinerClass(MaxTemperatureReducer.class);
        conf.setReducerClass(MaxTemperatureReducer.class);

        JobClient.runJob(conf);
    }
}

執行結果

上圖中，output目錄下的結果為未設定要使用的壓縮類，即Hadoop預設使用的是DEFLATE壓縮，副檔名為.deflate。output1目錄下的結果是設定使用Gzip壓縮。

需要注意

驗證過程中，需要注意的是，一定根據map和reduce方法的引數設定setOutputKeyClass和setOutputValueClass，否則會丟擲型別轉換異常，因為Hadoop預設的輸出Key和Value型別分別是LongWritable和Text。
本例中map和reduce輸出型別相同，如果不同，也可以單獨設定中間結果的輸出型別：

   conf.setMapOutputKeyClass(Text.class);
   conf.setMapOutputValueClass(IntWritable.class);

總結

要對MapReduce進行壓縮，需要把mapred.output.compress屬性設定為true，另外需要設定mapred.output.compression.codec屬性。如果要對map的輸出結果進行壓縮，則需要設定mapred.compress.map.output和mapred.map.output.compression.codec。