数据压缩

概述

这是mapreduce的一种优化策略：通过压缩编码对mapper或者reducer的输出进行压缩，以减少磁盘IO，提高MR程序运行速度（但相应增加了cpu运算负担）

1. Mapreduce支持将map输出的结果或者reduce输出的结果进行压缩，以减少网络IO或最终输出数据的体积
2. 压缩特性运用得当能提高性能，但运用不当也可能降低性能
3. 基本原则：
   运算密集型的job，少用压缩
   IO密集型的job，多用压缩

MR支持的压缩编码

为了支持多种压缩/解压缩算法,hadoop引入了编码/解码器,如下

压缩性能比较

压缩方式选择

Gzip压缩

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

缺点: 不支持split

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

Bzip2压缩

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

缺点: 压缩/解压速度慢,不支持native

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

Lzo压缩

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

缺点: 压缩率比gzip要低一些,hadoop本身不支持,需要安装.在应用中对lzo格式的文件需要做一些特殊处理(为了支持split需要建索引,还需要指定inoutformat为lzo格式)

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

Snappy压缩

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

缺点: 不支持split,压缩率比gzip要低,hadoop本身不支持,需要安装

应用场景: 当mapreduce作业的map输出的数据比较大的时候,作为map到reduce的中间数据的压缩格式,或者作为一个mapreduce作业的输出和另外一个mapreduce作业的输入

压缩位置选择

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

输入端采用压缩

在有大量数据并计划重复处理的情况下,应该考虑对输入进行压缩.然而,你无须显示指定使用的编解码方式.
hadoop自动检查文件扩展名.如果扩展名能够匹配,就会用恰当的编解码方式对文件进行压缩和解压,否则hadoop不会使用任何编解码器

输出采用压缩

当map任务输出的中间数据量很大时,应考虑在此阶段采用压缩技术.这能显著改善内部数据shuffle过程.shuffle是消耗资源最多的环节.可用于压缩mapper输出的快速编解码器包括LZO或者snapper

reducer输出采用压缩

在此阶段启用压缩技术能够减少要存储的数据量,因此降低所需的磁盘空间.
当mapreduce作业形成作业链条时,所以启用压缩同样有效

压缩配置参数

要在hadoop中启用压缩,可以配置如下参数(mapred-site.xml中)

代码

测试输出压缩

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;
import java.io.*;
public class TestCompress {
    public static void main(String[] args) throws IOException, ClassNotFoundException {
        //测试压缩
        compress("/Users/jdxia/Desktop/website/data/input/order.txt", "org.apache.hadoop.io.compress.BZip2Codec");
    }
    //测试压缩
    @SuppressWarnings({"resource", "unchecked"})
    private static void compress(String filename, String method) throws IOException, ClassNotFoundException {
        //获取输入流
        FileInputStream fis = new FileInputStream(new File(filename));
        Class className = Class.forName(method);
        CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(className, new Configuration());
        //获取输出流,输出的文件是文件名加后缀
        FileOutputStream fos = new FileOutputStream(new File(filename + codec.getDefaultExtension()));
        CompressionOutputStream cos = codec.createOutputStream(fos);
        //流的对拷,暂时不关闭流,最后在filename所在文件夹中会有个压缩文件
        IOUtils.copyBytes(fis, cos, 1024*1024*5, false);
        //关闭资源
        fis.close();
        cos.close();
        fos.close();
    }
}

测试输入压缩

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 java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
public class TestCompress {
    public static void main(String[] args) throws IOException, ClassNotFoundException {
        //测试压缩
        decompress("/Users/jdxia/Desktop/website/data/input/order.txt.bz2");
    }
    private static void decompress(String filename) throws IOException {
        //校验是否能解压缩
        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;
        }
        //获取输入流
        CompressionInputStream cis = codec.createInputStream(new FileInputStream(new File(filename)));
        //获取输出流
        FileOutputStream fos = new FileOutputStream(new File(filename + ".decoded"));
        //流的对拷
        IOUtils.copyBytes(cis, fos, 1024*1024*5, false);
        //关闭资源
        cis.close();
        fos.close();
    }
}

map输出采用压缩

即使你的MapReduce的输入输出文件都是未压缩的文件,你仍然可以对map任务的中间结果输出做压缩,因为他要写在硬盘并且通过网络传输到reduce节点,对其压缩可以提高很多性能,这些工作只要设置两个属性即可

这些工作只要设置两个属性就可以

map端输入可以根据扩展名来

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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
public class WorldCountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        //开启map端输出压缩
        conf.setBoolean("mapreduce.map.output.compress", true);
        conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);
        Job job = Job.getInstance(conf);
        //告诉框架,我们程序的位置
        job.setJarByClass(WorldCountDriver.class);
        //告诉框架,我们程序所用的mapper类和reduce类是什么
        job.setMapperClass(WorldCountMapper.class);
        job.setReducerClass(WorldCountReducer.class);
        //告诉框架我们程序输出的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        //告诉框架,我们程序使用的数据读取组件,结果输出所用的组件是什么
        //TextInputFormat是mapreduce程序中内置的一种读取数据组件,准备的叫做读取文本的输入组件
        job.setInputFormatClass(TextInputFormat.class);
        //job.setOutputFormatClass(TextOutputFormat.class);
        //告诉框架,我们要处理的数据文件在那个路径下
        FileInputFormat.setInputPaths(job,new Path("/Users/jdxia/Desktop/website/data/input"));
        //告诉框架我们的处理结果要输出到什么地方
        FileOutputFormat.setOutputPath(job,new Path("/Users/jdxia/Desktop/website/data/output"));
        //这边不用submit,因为一提交就和我这个没关系了,我这就断开了就看不见了
//        job.submit();
        //提交后,然后等待服务器端返回值,看是不是true
        boolean res = job.waitForCompletion(true);
        //设置成功就退出码为0
        System.exit(res?0:1);
    }
}

reduce输出采用压缩

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.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
public class WorldCountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        //告诉框架,我们程序的位置
        job.setJarByClass(WorldCountDriver.class);
        //告诉框架,我们程序所用的mapper类和reduce类是什么
        job.setMapperClass(WorldCountMapper.class);
        job.setReducerClass(WorldCountReducer.class);
        //告诉框架我们程序输出的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        //告诉框架,我们程序使用的数据读取组件,结果输出所用的组件是什么
        //TextInputFormat是mapreduce程序中内置的一种读取数据组件,准备的叫做读取文本的输入组件
        job.setInputFormatClass(TextInputFormat.class);
        //job.setOutputFormatClass(TextOutputFormat.class);
        //告诉框架,我们要处理的数据文件在那个路径下
        FileInputFormat.setInputPaths(job,new Path("/Users/jdxia/Desktop/website/data/input"));
        //告诉框架我们的处理结果要输出到什么地方
        FileOutputFormat.setOutputPath(job,new Path("/Users/jdxia/Desktop/website/data/output"));
        //设置reduce端输出压缩开启
        FileOutputFormat.setCompressOutput(job, true);
        //设置压缩的方式
        FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);
        //提交后,然后等待服务器端返回值,看是不是true
        boolean res = job.waitForCompletion(true);
        //设置成功就退出码为0
        System.exit(res?0:1);
    }
}

数据流的压缩和解压缩

CompressionCodec有两个方法可以用于轻松的压缩或解压缩数据.要想对正在被写入一个输出流的数据进行压缩,我们可以使用createOutputStream(OutputStreamout)方法创建一个CompressionOutputStream,将其以压缩格式写入底层的流.

相反,想要对从输入流读取而来的数据进行解压缩,则调用createInputStream(InputStream)函数,从而获得一个CompressionInputStream,从而从底层的流读取未压缩的数据

Reducer输出压缩

在配置参数或在代码中都可以设置reduce的输出压缩

1. 在配置参数中设置

mapreduce.output.fileoutputformat.compress=false
mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.DefaultCodec
mapreduce.output.fileoutputformat.compress.type=RECORD

1. 在代码中设置

Job job = Job.getInstance(conf);
FileOutputFormat.setCompressOutput(job, true);
FileOutputFormat.setOutputCompressorClass(job, (Class<? extends CompressionCodec>) Class.forName(""));

Mapper输出压缩

在配置参数或在代码中都可以设置reduce的输出压缩

1. 在配置参数中设置

mapreduce.map.output.compress=false
mapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.DefaultCodec

1. 在代码中设置：

conf.setBoolean(Job.MAP_OUTPUT_COMPRESS, true);
conf.setClass(Job.MAP_OUTPUT_COMPRESS_CODEC, GzipCodec.class, CompressionCodec.class);

压缩文件的读取(源码)

Hadoop自带的InputFormat类内置支持压缩文件的读取，比如TextInputformat类，在其initialize方法中：

public void initialize(InputSplit genericSplit, TaskAttemptContext context) throws IOException {
    FileSplit split = (FileSplit) genericSplit;
    Configuration job = context.getConfiguration();
    this.maxLineLength = job.getInt(MAX_LINE_LENGTH, Integer.MAX_VALUE);
    start = split.getStart();
    end = start + split.getLength();
    final Path file = split.getPath();
    // open the file and seek to the start of the split
    final FileSystem fs = file.getFileSystem(job);
    fileIn = fs.open(file);
    //根据文件后缀名创建相应压缩编码的codec
    CompressionCodec codec = new CompressionCodecFactory(job).getCodec(file);
    if (null!=codec) {
      isCompressedInput = true;    
      decompressor = CodecPool.getDecompressor(codec);
      //判断是否属于可切片压缩编码类型
      if (codec instanceof SplittableCompressionCodec) {
        final SplitCompressionInputStream cIn =
          ((SplittableCompressionCodec)codec).createInputStream(
            fileIn, decompressor, start, end,
            SplittableCompressionCodec.READ_MODE.BYBLOCK);
         //如果是可切片压缩编码，则创建一个CompressedSplitLineReader读取压缩数据
        in = new CompressedSplitLineReader(cIn, job,
            this.recordDelimiterBytes);
        start = cIn.getAdjustedStart();
        end = cIn.getAdjustedEnd();
        filePosition = cIn;
      } else {
        //如果是不可切片压缩编码，则创建一个SplitLineReader读取压缩数据，并将文件输入流转换成解压数据流传递给普通SplitLineReader读取
        in = new SplitLineReader(codec.createInputStream(fileIn,
            decompressor), job, this.recordDelimiterBytes);
        filePosition = fileIn;
      }
    } else {
      fileIn.seek(start);
       //如果不是压缩文件，则创建普通SplitLineReader读取数据
      in = new SplitLineReader(fileIn, job, this.recordDelimiterBytes);
      filePosition = fileIn;
    }