尚硅谷大数据技术之Hadoop（MapReduce）第4章 Hadoop数据压缩

2019年6月8日 229次阅读来源: 尚硅谷教育

4.1 概述
压缩技术能够有效减少底层存储系统（HDFS）读写字节数。压缩提高了网络带宽和磁盘空间的效率。在Hadoop下，尤其是数据规模很大和工作负载密集的情况下，使用数据压缩显得非常重要。在这种情况下，I/O操作和网络数据传输要花大量的时间。还有，Shuffle与Merge过程同样也面临着巨大的I/O压力。
鉴于磁盘I/O和网络带宽是Hadoop的宝贵资源，数据压缩对于节省资源、最小化磁盘I/O和网络传输非常有帮助。不过，尽管压缩与解压操作的CPU开销不高，其性能的提升和资源的节省并非没有代价。
如果磁盘I/O和网络带宽影响了MapReduce作业性能，在任意MapReduce阶段启用压缩都可以改善端到端处理时间并减少I/O和网络流量。
压缩Mapreduce的一种优化策略：通过压缩编码对Mapper或者Reducer的输出进行压缩，以减少磁盘IO，提高MR程序运行速度（但相应增加了cpu运算负担）。
注意：压缩特性运用得当能提高性能，但运用不当也可能降低性能。
基本原则：
（1）运算密集型的job，少用压缩
（2）IO密集型的job，多用压缩
4.2 MR支持的压缩编码
压缩格式 hadoop自带？算法文件扩展名是否可切分换成压缩格式后，原来的程序是否需要修改
DEFAULT 是，直接使用 DEFAULT .deflate 否和文本处理一样，不需要修改
Gzip 是，直接使用 DEFAULT .gz 否和文本处理一样，不需要修改
bzip2 是，直接使用 bzip2 .bz2 是和文本处理一样，不需要修改
LZO 否，需要安装 LZO .lzo 是需要建索引，还需要指定输入格式
Snappy 否，需要安装 Snappy .snappy 否和文本处理一样，不需要修改
为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器，如下表所示
压缩格式对应的编码/解码器
DEFLATE org.apache.hadoop.io.compress.DefaultCodec
gzip org.apache.hadoop.io.compress.GzipCodec
bzip2 org.apache.hadoop.io.compress.BZip2Codec
LZO com.hadoop.compression.lzo.LzopCodec
Snappy org.apache.hadoop.io.compress.SnappyCodec
压缩性能的比较
压缩算法原始文件大小压缩文件大小压缩速度解压速度
gzip 8.3GB 1.8GB 17.5MB/s 58MB/s
bzip2 8.3GB 1.1GB 2.4MB/s 9.5MB/s
LZO 8.3GB 2.9GB 49.3MB/s 74.6MB/s
<u>http://google.github.io/snappy/</u>

On a single core of a Core i7 processor in 64-bit mode, Snappy compressesat about 250 MB/sec or more and decompresses at about 500 MB/sec or more.

4.3 压缩方式选择

4.3.1 Gzip压缩

优点：压缩率比较高，而且压缩/解压速度也比较快；hadoop本身支持，在应用中处理gzip格式的文件就和直接处理文本一样；大部分linux系统都自带gzip命令，使用方便。

缺点：不支持split。

应用场景：当每个文件压缩之后在130M以内的（1个块大小内），都可以考虑用gzip压缩格式。例如说一天或者一个小时的日志压缩成一个gzip文件，运行mapreduce程序的时候通过多个gzip文件达到并发。hive程序，streaming程序，和java写的mapreduce程序完全和文本处理一样，压缩之后原来的程序不需要做任何修改。

4.3.2 Bzip2压缩

优点：支持split；具有很高的压缩率，比gzip压缩率都高；hadoop本身支持，但不支持native；在linux系统下自带bzip2命令，使用方便。

缺点：压缩/解压速度慢；不支持native。

应用场景：适合对速度要求不高，但需要较高的压缩率的时候，可以作为mapreduce作业的输出格式；或者输出之后的数据比较大，处理之后的数据需要压缩存档减少磁盘空间并且以后数据用得比较少的情况；或者对单个很大的文本文件想压缩减少存储空间，同时又需要支持split，而且兼容之前的应用程序（即应用程序不需要修改）的情况。

4.3.3 Lzo压缩

优点：压缩/解压速度也比较快，合理的压缩率；支持split，是hadoop中最流行的压缩格式；可以在linux系统下安装lzop命令，使用方便。

缺点：压缩率比gzip要低一些；hadoop本身不支持，需要安装；在应用中对lzo格式的文件需要做一些特殊处理（为了支持split需要建索引，还需要指定inputformat为lzo格式）。

应用场景：一个很大的文本文件，压缩之后还大于200M以上的可以考虑，而且单个文件越大，lzo优点越越明显。

4.3.4 Snappy压缩

优点：高速压缩速度和合理的压缩率。

缺点：不支持split；压缩率比gzip要低；hadoop本身不支持，需要安装；

应用场景：当Mapreduce作业的Map输出的数据比较大的时候，作为Map到Reduce的中间数据的压缩格式；或者作为一个Mapreduce作业的输出和另外一个Mapreduce作业的输入。

4.4压缩位置选择

压缩可以在MapReduce作用的任意阶段启用。

《尚硅谷大数据技术之Hadoop（MapReduce）第4章 Hadoop数据压缩》 image.png

4.5 压缩参数配置

要在Hadoop中启用压缩，可以配置如下参数：

参数默认值阶段建议

io.compression.codecs

（在core-site.xml中配置） org.apache.hadoop.io.compress.DefaultCodec, org.apache.hadoop.io.compress.GzipCodec, org.apache.hadoop.io.compress.BZip2Codec

输入压缩 Hadoop使用文件扩展名判断是否支持某种编解码器

mapreduce.map.output.compress（在mapred-site.xml中配置） false mapper输出这个参数设为true启用压缩

mapreduce.map.output.compress.codec（在mapred-site.xml中配置） org.apache.hadoop.io.compress.DefaultCodec mapper输出使用LZO或snappy编解码器在此阶段压缩数据

mapreduce.output.fileoutputformat.compress（在mapred-site.xml中配置） false reducer输出这个参数设为true启用压缩

mapreduce.output.fileoutputformat.compress.codec（在mapred-site.xml中配置） org.apache.hadoop.io.compress. DefaultCodec reducer输出使用标准工具或者编解码器，如gzip和bzip2

mapreduce.output.fileoutputformat.compress.type（在mapred-site.xml中配置） RECORD reducer输出 SequenceFile输出使用的压缩类型：NONE和BLOCK

4.6 压缩实操案例

4.6.1 数据流的压缩和解压缩

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.atguigu.mapreduce.compress;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
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;

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

// 压缩
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();
}

// 解压缩
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();
}

}
4.6.2 Map输出端采用压缩
即使你的MapReduce的输入输出文件都是未压缩的文件，你仍然可以对map任务的中间结果输出做压缩，因为它要写在硬盘并且通过网络传输到reduce节点，对其压缩可以提高很多性能，这些工作只要设置两个属性即可，我们来看下代码怎么设置：
1）给大家提供的hadoop源码支持的压缩格式有：BZip2Codec 、DefaultCodec
package com.atguigu.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.atguigu.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 {
    // 1 获取一行
    String line = value.toString();
    // 2 切割
    String[] words = line.split(" ");
    // 3 循环写出
    for(String word:words){
        context.write(new Text(word), new IntWritable(1));
    }
}

}
3）Reducer保持不变
package com.atguigu.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;
    // 1 汇总
    for(IntWritable value:values){
        count += value.get();
    }
    
    // 2 输出
    context.write(key, new IntWritable(count));
}

}
4.6.3 Reduce输出端采用压缩
基于workcount案例处理
1）修改驱动
package com.atguigu.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保持不变（详见4.6.2）

本教程由尚硅谷教育大数据研究院出品，如需转载请注明来源，欢迎大家关注尚硅谷公众号（atguigu）了解更多。

    原文作者：尚硅谷教育
    原文地址: https://www.jianshu.com/p/6f0482fc8200
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。

4****.****3 压缩****方式****选择

4.3****.****1**** Gzip压缩

4.3.2 B****zip2压缩

4.3.3 L****zo压缩

4.3.4 S****nappy压缩

4.3 压缩方式选择

4.3.1 Gzip压缩

4.3.2 Bzip2压缩

4.3.3 Lzo压缩

4.3.4 Snappy压缩