Hadoop数据压缩、MR支持的压缩编码、压缩方式选择、压缩位置选择、压缩实操案例、Map输出端采用压缩、Reduce输出端采用压缩、Gzip压缩、Bzip2压缩、Lzo压缩、Snappy压缩_mr自带的压缩编码

14. Hadoop数据压缩

14.1 概述

14.1.1 压缩的好处和坏处

压缩的优点：以减少磁盘IO、减少磁盘存储空间。
压缩的缺点：增加CPU开销。

14.1.2 压缩原则

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

14.2 MR支持的压缩编码

14.2.1 压缩算法对比介绍

14.2.2 压缩性能的比较

http://google.github.io/snappy/
Snappy is a compression/decompression library. It does not aim for maximum compression, or compatibility with any other compression library; instead, it aims for very high speeds and reasonable compression. For instance, compared to the fastest mode of zlib, Snappy is an order of magnitude faster for most inputs, but the resulting compressed files are anywhere from 20% to 100% bigger.On a single core of a Core i7 processor in 64-bit mode, Snappy compresses at about 250 MB/sec or more and decompresses at about 500 MB/sec or more.

14.3 压缩方式选择

  压缩方式选择时重点考虑：压缩/解压缩速度、压缩率（压缩后存储大小）、压缩后是否可以支持切片。

14.3.1 Gzip压缩

优点：压缩率比较高；
缺点：不支持Split；压缩/解压速度一般；

14.3.2 Bzip2压缩

优点：压缩率高；支持Split；
缺点：压缩/解压速度慢。

14.3.3 Lzo压缩

优点：压缩/解压速度比较快；支持Split；
缺点：压缩率一般；想支持切片需要额外创建索引。

14.3.4 Snappy压缩

优点：压缩和解压缩速度快；
缺点：不支持Split；压缩率一般；

14.3.5 压缩位置选择

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

14.4 压缩位置选择

14.4.1 为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器

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

hadoop checknative命令

  Snappy是和centos7.5和hadoop3.0结合使用，他是支持压缩的，如果在本地idea里面，他没有这个运行环境，他这个Snappy是不支持的

14.5 压缩实操案例

14.5.1 Map输出端采用压缩

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

14.5.1.1 给大家提供的Hadoop源码支持的压缩格式有：BZip2Codec、DefaultCodec

将之前workcount包里面的代码同时复制到compress包里面，然后再Driver里面添加两行代码

// 开启map端输出压缩
        conf.setBoolean("mapreduce.map.output.compress", true);

        // 设置map端输出压缩方式
        conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

1
2
3
4
5
6

14.5.1.2 编辑WorkCountDriver类

package com.summer.mapreduce.compress;

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.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * @author Redamancy
 * @create 2022-10-08 17:15
 */
public class WordCountDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        // 输入输出路径需要根据自己电脑上实际的输入输出路径设置
        args = new String[] {"D:\\Acode\\Hadoop\\input\\inputhello", "D:\\Acode\\Hadoop\\output\\output1"};

        //1 获取job
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        // 开启map端输出压缩
        conf.setBoolean("mapreduce.map.output.compress", true);

        // 设置map端输出压缩方式
        conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

        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 ? 0 : 1);

    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

14.5.1.3 编写WorkCountMapper类

package com.summer.mapreduce.compress;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author Redamancy
 * @create 2022-10-08 16:47
 */


/**
 * KEYIN, map阶段输入的key的类型：LongWritable
 * VALUEIN, map阶段输入的value的类型：Text
 * KEYOUT, map阶段输出的key的类型：Text
 * VALUEOUT,map阶段输出的kvalue的类型：IntWritable
 */
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    private Text outK = new Text();
    IntWritable outV = new IntWritable(1);
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {

        //1 获取一行
        //wo ni
        String line = value.toString();

        //2 切割
        //wo
        //ni
        String[] words = line.split(" ");

        for (String word : words) {

            //封装outK
            outK.set(word);

            //写出
            context.write(outK, outV);
        }
    }
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

14.5.1.4 编写WorkCountReducer类

package com.summer.mapreduce.compress;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * @author Redamancy
 * @create 2022-10-08 17:08
 */

/**
 * KEYIN, reduce阶段输入的key的类型：Text
 * VALUEIN, reduce阶段输入的value的类型：IntWritable
 * KEYOUT, reduce阶段输出的key的类型：Text
 * VALUEOUT,reduce阶段输出的kvalue的类型：IntWritable
 */
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable outV = new IntWritable();
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {

        int sum = 0;
        //ha(1,1)
        //累加
        for (IntWritable value : values) {
            sum += value.get();
        }

        outV.set(sum);

        //写出
        context.write(key, outV);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

14.5.1.5 运行结果

这个是在map输出阶段进行压缩文件，最终的输出没有进行压缩，所以这里看不到文件压缩

14.5.2 Reduce输出端采用压缩

基于WordCount案例处理。

14.5.2.1 修改驱动

 // 设置reduce端输出压缩开启
        FileOutputFormat.setCompressOutput(job, true);

        // 设置压缩的方式
        FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);
//        FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
//        FileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class);
1
2
3
4
5
6
7

在上面的基础上添加这两块代码，压缩方式可以从这三个里面选一个，DefaultCodec压缩在windows系统上会出现乱码，因为没有这种编码格式，其他两种都可以正常看到

14.5.2.2 Mapper和Reducer保持不变(详细见14.5.1.2，14.5.1.3)

14.5.2.3 运行结果

是一个压缩包，后缀是bz2

结果和上面一样，只不过将输出的结果压缩了，其他两种也可以试试

14.5.2.4 将map阶段压缩换成Snappy

  Snappy是和centos7.5和hadoop3.0结合使用，他是支持压缩的，如果在本地idea里面，他没有这个运行环境，他这个Snappy是不支持的