简洁而不简单

Hadoop 数据压缩

数据压缩优点和缺点

​ 压缩技术能够 有效减少底层存储系统（HDFS）读写字节数 。压缩提高了网络带宽和磁盘空间的效率。在 Hadoop 下，尤其是数据规模很大和工作负载密集的情况下，使用数据压缩显得非常重要。在这种情况下，IO 操作和网络数据传输要花大量的时间。还有，Shuffle 与 Merge 过程同样也面临着巨大的 IO 压力鳘于磁盘 IO 和网络带宽是 Hadoop 的宝贵资源， 数据压缩对于节省资源、最小化磁盘 IO 和网络传输非常有帮助。

​ 不过，尽管压缩与解压操作的 CPU 开销不髙，其性能的提升和资源的节省并非没有代价。如果磁盘 IO 和网络带宽影响了 MapReduce 作业性能，在任意 MapReduce 阶段启用压缩都可以改善端到端处理时间并減少 IO 和网络流量。

压缩策略和原则

​ 压缩是提高 Hadoop 运行效率的一种优化策略通过对 Mapper、Reducer 运行过程的数据进行压缩，以减少磁盘 IO，提高 MR 程序运行速度 。
​ 注意：釆用压缩技术减少了磁盘 IO，但同时 增加了 CPU 运算负担。所以，压缩特性运用得当能提高性能，但运用不当也可能降低性能压缩基本原则：

（1）运算密集型的 job，少用压缩
（2）IO 密集型的 job，多用压缩！！

MR 支持的压缩编码

为了支持多种压缩 / 解压缩算法，Hadoop 引入了编码 / 解码器，如下表所示。

压缩性能的比较

压缩方式选择

Gzip 压缩

Bzip2 压缩

Lzo 压缩

Snappy 压缩

压缩位置选择

压缩参数配置

压缩案例

public class TestCompress {public static void main(String[] args) throws Exception {compress("e:/hello.txt","org.apache.hadoop.io.compress.BZip2Codec");
//        decompress("e:/hello.txt.bz2");
    }

    // 1、压缩
    private static void compress(String filename, String method) throws Exception {
        
        //（1）获取输入流
        FileInputStream fis = new FileInputStream(new File(filename));
        
        Class codecClass = Class.forName(method);
        
        CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, new Configuration());
        
        //（2）获取输出流
        FileOutputStream fos = new FileOutputStream(new File(filename +codec.getDefaultExtension()));
        CompressionOutputStream cos = codec.createOutputStream(fos);
        
        //（3）流的对拷
        IOUtils.copyBytes(fis, cos, 1024*1024*5, false);
        
        //（4）关闭资源
        fis.close();
        cos.close();
        fos.close();}

    // 2、解压缩
    private static void decompress(String filename) throws FileNotFoundException, IOException {
        
        //（0）校验是否能解压缩
        CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());

        CompressionCodec codec = factory.getCodec(new Path(filename));
        
        if (codec == null) {System.out.println("cannot find codec for file" + filename);
            return;
        }
        
        //（1）获取输入流
        CompressionInputStream cis = codec.createInputStream(new FileInputStream(new File(filename)));
        
        //（2）获取输出流
        FileOutputStream fos = new FileOutputStream(new File(filename + ".decoded"));
        
        //（3）流的对拷
        IOUtils.copyBytes(cis, fos, 1024*1024*5, false);
        
        //（4）关闭资源
        cis.close();
        fos.close();}
}

Map 输出端采用压缩

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);
    }
}

Mapper 和 Reducer 代码不变

Reduce 输出端采用压缩

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);
    }
}

相关资料

本文配套GitHub：https://github.com/zhutiansam…