Hadoop案例（二）壓縮解壓縮

壓縮/解壓縮案例

一. 對數據流的壓縮和解壓縮

CompressionCodec有兩個方法可以用於輕松地壓縮或解壓縮數據。要想對正在被寫入一個輸出流的數據進行壓縮，我們可以使用createOutputStream(OutputStreamout)方法創建一個CompressionOutputStream，將其以壓縮格式寫入底層的流。相反，要想對從輸入流讀取而來的數據進行解壓縮，則調用createInputStream(InputStreamin)函數，從而獲得一個CompressionInputStream，從而從底層的流讀取未壓縮的數據。

測試一下如下壓縮方式：

DEFLATE	org.apache.hadoop.io.compress.DefaultCodec
gzip	org.apache.hadoop.io.compress.GzipCodec
bzip2	org.apache.hadoop.io.compress.BZip2Codec

package com.xyg.mapreduce.compress;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
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.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;

public class TestCompress {
    
    public static void main(String[] args) throws Exception, IOException {
//        compress("e:/test.txt","org.apache.hadoop.io.compress.BZip2Codec");
        decompres("e:/test.txt.bz2");
    }
    
    /*
     * 壓縮
     * filername：要壓縮文件的路徑
     * method：欲使用的壓縮的方法（org.apache.hadoop.io.compress.BZip2Codec）
     */
    public static void compress(String filername, String method) throws ClassNotFoundException, IOException {
        
        // 1 創建壓縮文件路徑的輸入流
        File fileIn = new File(filername);
        InputStream in = new FileInputStream(fileIn);
        
        // 2 獲取壓縮的方式的類
        Class codecClass = Class.forName(method);
        
        Configuration conf = new Configuration();
        // 3 通過名稱找到對應的編碼/解碼器
        CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);

        // 4 該壓縮方法對應的文件擴展名
        File fileOut = new File(filername + codec.getDefaultExtension());

        OutputStream out = new FileOutputStream(fileOut);
        CompressionOutputStream cout = codec.createOutputStream(out);

        // 5 流對接
        IOUtils.copyBytes(in, cout, 1024 * 1024 * 5, false); // 緩沖區設為5MB

        // 6 關閉資源
        in.close();
        cout.close();
        out.close();
    }

    /*
     * 解壓縮
     * filename：希望解壓的文件路徑
     */
    public static void decompres(String filename) throws FileNotFoundException, IOException {

        Configuration conf = new Configuration();
        CompressionCodecFactory factory = new CompressionCodecFactory(conf);
        
        // 1 獲取文件的壓縮方法
        CompressionCodec codec = factory.getCodec(new Path(filename));
        
        // 2 判斷該壓縮方法是否存在
        if (null == codec) {
            System.out.println("Cannot find codec for file " + filename);
            return;
        }

        // 3 創建壓縮文件的輸入流
        InputStream cin = codec.createInputStream(new FileInputStream(filename));
        
        // 4 創建解壓縮文件的輸出流
        File fout = new File(filename + ".decoded");
        OutputStream out = new FileOutputStream(fout);

        // 5 流對接
        IOUtils.copyBytes(cin, out, 1024 * 1024 * 5, false);

        // 6 關閉資源
        cin.close();
        out.close();
    }
}

二. 在Map輸出端采用壓縮

即使你的MapReduce的輸入輸出文件都是未壓縮的文件，你仍然可以對map任務的中間結果輸出做壓縮，因為它要寫在硬盤並且通過網絡傳輸到reduce節點，對其壓縮可以提高很多性能，這些工作只要設置兩個屬性即可，我們來看下代碼怎么設置：

給大家提供的hadoop源碼支持的壓縮格式有：BZip2Codec 、DefaultCodec

package com.xyg.mapreduce.compress;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCountDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        Configuration configuration = new Configuration();

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

        Job job = Job.getInstance(configuration);

        job.setJarByClass(WordCountDriver.class);

        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

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

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

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        boolean result = job.waitForCompletion(true);

        System.exit(result ? 1 : 0);
    }
}

2）Mapper保持不變

package com.xyg.mapreduce.compress;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
    
    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        
        String line = value.toString();
        
        String[] words = line.split(" ");
        
        for(String word:words){
            context.write(new Text(word), new IntWritable(1));
        }
    }
}

3）Reducer保持不變

package com.xyg.mapreduce.compress;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
    
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values,
            Context context) throws IOException, InterruptedException {
        
        int count = 0;
        
        for(IntWritable value:values){
            count += value.get();
        }
        
        context.write(key, new IntWritable(count));
    }
}

三. 在Reduce輸出端采用壓縮

基於workcount案例處理

1）修改驅動

package com.xyg.mapreduce.compress;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.io.compress.Lz4Codec;
import org.apache.hadoop.io.compress.SnappyCodec;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCountDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        
        Configuration configuration = new Configuration();
        
        Job job = Job.getInstance(configuration);
        
        job.setJarByClass(WordCountDriver.class);
        
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        // 設置reduce端輸出壓縮開啟
        FileOutputFormat.setCompressOutput(job, true);
        
        // 設置壓縮的方式
        FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class); 
//        FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); 
//        FileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class); 
        
        boolean result = job.waitForCompletion(true);
        
        System.exit(result?1:0);
    }
}

2）Mapper和Reducer保持不變