Hadoop-数据压缩

清疚 2022-10-19 00:43 184阅读 0赞

# 数据压缩 #

## 概述 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]

计算密集型

计算密集型任务的特点是要进行大量的计算，消耗CPU资源，比如计算圆周率、对视频进行高清解码等等，全靠CPU的运算能力。这种计算密集型任务虽然也可以用多任务完成，但是任务越多，花在任务切换的时间就越多，CPU执行任务的效率就越低，所以，要最高效地利用CPU，计算密集型任务同时进行的数量应当等于CPU的核心数。计算密集型任务由于主要消耗CPU资源，因此，代码运行效率至关重要。

IO 密集型

IO密集型，涉及到网络、磁盘IO的任务都是IO密集型任务，这类任务的特点是CPU消耗很少，任务的大部分时间都在等待IO操作完成（因为IO的速度远远低于CPU和内存的速度）。对于IO密集型任务，任务越多，CPU效率越高，但也有一个限度。常见的大部分任务都是IO密集型任务，比如Web应用，上传下载等。IO密集型任务执行期间，99%的时间都花在IO上，花在CPU上的时间很少。

### MR支持的压缩编码 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]

其中的可切分的意思是，压缩完成之后是否可以对文件进行切分。

Hadoop引入的编解码器如下：

![在这里插入图片描述][20210708170031243.png]

压缩的性能比较：

![在这里插入图片描述][20210708170038235.png]

在 64 位模式的 Core i7 处理器的单核上，Snappy 以大约 250 MB/秒或更多的时间进行压缩，并以大约 500 MB/秒或更多的速度进行解压。

压缩性能总结：压缩文件越大，压缩/解压文件越快。

GZip  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 2]

BZip2  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 3]

LZO（拉泽罗）

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 4]

Snappy

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 5]

### 压缩位置选择 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 6]

能对数据进行压缩的阶段：Map之前，Map之后Reduce之前以及Reduce之后；而Map之前最好是使用能切分的压缩方式，因为不可切分的压缩文件如果太大的话对于Map处理非常耗时。

### 压缩参数配置 ###

<table> 
 <thead> 
  <tr> 
   <th>阶段</th> 
   <th>key</th> 
   <th>value</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>输入压缩</td> 
   <td>参数</td> 
   <td>io.compression.codecs（core-site.xml）</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>默认值</td> 
   <td>“”</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>备注</td> 
   <td>Hadoop使用文件扩展名判断是否支持某种编解码器；要能解压那些文件，需要配置对应的编解码器</td> 
  </tr> 
  <tr> 
   <td>mapper输出</td> 
   <td>参数</td> 
   <td>mapreduce.map.output.compress（mapred-site.xml）</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>默认</td> 
   <td>false</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>备注</td> 
   <td>这个参数设置true启用压缩</td> 
  </tr> 
  <tr> 
   <td>mapper输出</td> 
   <td>参数</td> 
   <td>mapreduce.map.output.compress.codec（mapred-site.xml）</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>默认</td> 
   <td>org.apache.hadoop.io.compress.DefaultCodec</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>备注</td> 
   <td>企业多使用LZO或者Snappy编解码器在此阶段压缩数据</td> 
  </tr> 
  <tr> 
   <td>reducer输出</td> 
   <td>参数</td> 
   <td>mapreduce.output.fileoutputformat.compress（mapred-site.xml）</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>默认</td> 
   <td>false</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>备注</td> 
   <td>这个参数设置true启用压缩</td> 
  </tr> 
  <tr> 
   <td>reducer输出</td> 
   <td>参数</td> 
   <td>mapreduce.output.fileoutputformat.compress.codec（mapred-site.xml）</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>默认</td> 
   <td>org.apache.hadoop.io.compress.DefaultCodec</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>备注</td> 
   <td>使用标准工具或者编解码器,如gzip和bzip2</td> 
  </tr> 
  <tr> 
   <td>reducer输出</td> 
   <td>参数</td> 
   <td>mapreduce.output.fileoutputformat.compress.type（mapred-site.xml）</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>默认</td> 
   <td>RECORD</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>备注</td> 
   <td>SequenceFile输出使用的压缩类型 :NONE和BLOCK（块），默认按行压缩</td> 
  </tr> 
 </tbody> 
</table>

### 案例实操 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 7]

压缩测试

![在这里插入图片描述][20210708170202937.png]

public class TestCompress { 
    
        public static void main (String[] args) throws Exception { 
    
            // 压缩Bzip2
            compress("/home/lxj/hadoop-data/input/compressTest/rootLogger.log",
                    "org.apache.hadoop.io.compress.BZip2Codec");
    
            // 压缩Gzip
            compress("/home/lxj/hadoop-data/input/compressTest/rootLogger.log",
                    "org.apache.hadoop.io.compress.GzipCodec");
    
            // 压缩Default
            compress("/home/lxj/hadoop-data/input/compressTest/rootLogger.log",
                    "org.apache.hadoop.io.compress.DefaultCodec");
    
            decompress("/home/lxj/hadoop-data/input/compressTest/rootLogger.log.bz2");
    
            decompress("/home/lxj/hadoop-data/input/compressTest/rootLogger.log.gz");
    
            decompress("/home/lxj/hadoop-data/input/compressTest/rootLogger.log.deflate");
    
        }
    
        /** * 压缩 * @param path * @param method * @throws Exception */
        private static void compress (String path, String method) throws Exception { 
    
            // 1、获取输入流
            FileInputStream fis = new FileInputStream(new File(path));
    
            // 2、获取输出流
            // 获取压缩编码器类
            Class classCodec = Class.forName(method);
            CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(classCodec, new Configuration());
    
            // 普通输出流
            FileOutputStream fos = new FileOutputStream(new File(path + codec.getDefaultExtension()));
    
            // 压缩输出流
            CompressionOutputStream cos = codec.createOutputStream(fos);
    
            // 3、流的对拷
            IOUtils.copyBytes(fis, cos, 1024 * 1024, false);
    
            // 4、关闭资源
            IOUtils.closeStream(cos);
            IOUtils.closeStream(fos);
            IOUtils.closeStream(fis);
        }
    
        /** * 解压缩 * @param path */
        private static void decompress (String path) throws Exception { 
    
            // 1、压缩方式检查
            CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());
            CompressionCodec codec = factory.getCodec(new Path(path));
    
            if (codec == null) { 
                System.out.println("can not compress");
                return;
            }
    
            // 2、获取输入流
            FileInputStream fis = new FileInputStream(new File(path));
            CompressionInputStream cis = codec.createInputStream(fis);
    
            // 3、获取输出流
            FileOutputStream fos = new FileOutputStream(new File(path + ".decode"));
    
            // 4、流的对拷
            IOUtils.copyBytes(cis, fos, 1024 * 1024, false);
    
            // 5、关闭资源
            IOUtils.closeStream(fos);
            IOUtils.closeStream(cis);
            IOUtils.closeStream(fis);
        }
    
    }

压缩和解压的结果如下：

![在这里插入图片描述][20210708170215654.png]

### Map输出阶段采用压缩 ###

即使你的 MapReduce 的输入输出文件都是未压缩的文件,你仍然可以对 Map 任务的中间结果输出做压缩,因为它要写在硬盘并且通过网络传输到 Reduce 节点,对其压缩可以提高很多性能,这些工作只要设置两个属性即可,我们来看下代码怎么设置。

只需要在Driver层指定以下参数即可

// 设置Mapper输出压缩
    conf.setBoolean("mapreduce.map.output.compress", true);
    // 设置压缩方法为bzip2
    conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

只设置Map输出压缩，不会影响最终输出的文件。

### Reduce输出阶段采用压缩 ###

// 设置开启reduce输出压缩
    FileOutputFormat.setCompressOutput(job, true);
    // 设置reduce输出压缩为bzip2
    FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);

输出结果变成了bzip2压缩文件

![在这里插入图片描述][20210708170002297.png]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]: /images/20221005/2a393db0c65341abae2102a6e604fe87.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]: /images/20221005/93bfd864c73946d2a363864aa301db82.png
[20210708170031243.png]: /images/20221005/bd64fec8a4254736a456c26666e98025.png
[20210708170038235.png]: /images/20221005/ead7134fbbba4be6bb1738b3c60237ae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 2]: /images/20221005/f1868e176ac244d2b908109da18d63c5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 3]: /images/20221005/f9bad87bbd0a4cff8a708ec431496a5e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 4]: /images/20221005/d8f1505f236b4860b59ff96b9418846f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 5]: /images/20221005/9115c220bd0b424182bf9558a265183e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 6]: /images/20221005/d1938bd8b06b401982ca4e41e8706804.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 7]: /images/20221005/f91d43a163fa4cf29aae8196d679e7cc.png
[20210708170202937.png]: /images/20221005/2ad2fe0726974a04815c2b59bf5ec031.png
[20210708170215654.png]: /images/20221005/994aaf1df1834405a389575278df9394.png
[20210708170002297.png]: /images/20221005/fcf40248e39e43be9f903f4f715724f0.png