hadoop 压缩框架

[toc]

hadoop 常见的压缩

| 格式 | split | native |压缩率 |速度 | 是否hadoop自带 |linux命令 | 换成压缩格式后，原来的应用程序是否要修改 | | :-------- | :--------:| :------: |:--------:| :------: |:--------:| :------: | |gzip | 否| 是 |很高| 比较快| 是，直接使用| 有 |和文本处理一样，不需要修改| |lzo| 是|是 |比较高 |很快| 否，需要安装 |有| 需要建索引，还需要指定输入格式| |snappy|否| 是 |比较高 |很快| 否，需要安装| 没有| 和文本处理一样，不需要修改| |bzip2 |是 |否 |最高 |慢 | 是，直接使用 |有 |和文本处理一样，不需要修改|

压缩使用例子

public class CompressTest {
 public static void main(String[] args) throws Exception {
     //要压缩的文件
  String name = "/Users/zwf/Downloads/BEH 7.0产品介绍-20161226.pptx";
  /**
  * 指定压缩方式可用的有：
  * 1. org.apache.hadoop.io.compress.DefaultCodec
  * 2. org.apache.hadoop.io.compress.GzipCodec
  * 3. org.apache.hadoop.io.compress.Bzip2Codec
  * 4. org.apache.hadoop.io.compress.LzopCodec
  * 5. org.apache.hadoop.io.compress.SnappyCodec
  **/
  compress("org.apache.hadoop.io.compress.DefaultCodec", name);
  decompress(name+".deflate");
 }
 //压缩过程
 public static void compress(String method ,String filename) throws Exception {

  File inFile = new File(filename);
  System.out.println(inFile.getTotalSpace());
  InputStream in = new FileInputStream(inFile);
  Configuration conf = new Configuration();
  //加载类，反射获取实例
  CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(Class.forName(method), conf);
  File outFile = new File(filename + codec.getDefaultExtension());
  CompressionOutputStream cout = codec.createOutputStream(new FileOutputStream(outFile));
  IOUtils.copyBytes(in,cout,1000,false);
  in.close();
  cout.close();
  System.out.println(outFile.getTotalSpace()/1024/1024);
 }
 //解压缩过程
 public static void decompress(String filename) throws Exception {
  
  File inFile = new File(filename);
  CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());
  //根据文件名后缀获取对应的压缩方式
  CompressionCodec codec = factory.getCodec(new Path(filename));
  if(codec == null) {
   System.out.println("connot found codec for file " + filename);
   return;
  }
  InputStream in = codec.createInputStream(new FileInputStream(inFile));
  File outFile = new File(filename + ".txt");
  OutputStream out = new FileOutputStream(outFile);
  IOUtils.copyBytes(in,out,1000,false);
  in.close();
  out.close();
  
  System.out.println(outFile.getTotalSpace()/1024/1024);
 }
 
}

主要类

CompressionCodec

主要功能：定义了压缩，和解压缩调用的方法主要实现： Lz4Codec ，GzipCodec，SnappyCodec ，BZip2Codec等主要方法:

    // 在底层输出流 out 的基础上创建 对应压缩算法的压缩流 CompressionOutputStream  象
    CompressionOutputStream createOutputStream(OutputStream out) throws IOException;
    CompressionOutputStream createOutputStream(OutputStream out, Compressor compressor) throws IOException；
    //获取本地压缩是否可用，返回压缩实现类
    Class<? extends Compressor> getCompressorType()
    Class<? extends Decompressor> getDecompressorType()
    //获取解压缩使用的输入流
    CompressionInputStream createInputStream(InputStream in) throws IOException;
    CompressionInputStream createInputStream(InputStream in,  Decompressor decompressor) throws IOException;
    // 创建压缩算法对应的压缩器
    Compressor createCompressor();
    Decompressor createDecompressor();
    //获取压缩算法对应的文件名后缀
    String getDefaultExtension();

CompressionCodecFactory

主要功能： CompressionCodec 的工厂，构造函数中初始化了一下存储现有压缩算法的实例，主要保存了

    {
       2zb.: org.apache.hadoop.io.compress.BZip2Codec,
       etalfed.: org.apache.hadoop.io.compress.DeflateCodec,
       yppans.: org.apache.hadoop.io.compress.SnappyCodec,
       zg.: org.apache.hadoop.io.compress.GzipCodec
    }

Compressor

主要功能：定义了压缩使用的接口方法，提供数据压缩功能。不同压缩都有对应的实现主要方法：

    //把数据放到压缩其中， off 指定从数字b的开始位置， len指定读取数据长度
    public void setInput(byte[] b, int off, int len);
    //判断压缩器是否能继续输入，如果不能表示需要开始压缩数据，
    public boolean needsInput();

    public void setDictionary(byte[] b, int off, int len);
    //返回已经压缩了的数据压缩前长度
    public long getBytesRead();
    
    //返回已经压缩了的数据压缩后长度
    public long getBytesWritten();
    //结束数据输入的过程
    public void finish();
   //判断压缩器中是否还有未压缩数据
    public boolean finished();
    //开始压缩数据
    public int compress(byte[] b, int off, int len) throws IOException;
    //重置压缩器
    public void reset();
   //关闭压缩器
    public void end();
    //方法更进一步允许使用 Hadoop 的配置系统， 重置并重新配置压缩 。
    public void reinit(Configuration conf);

抽象类CompressionOutputStream

主要作用：定义压缩输出流的接口

CompressorStream

主要作用：持有一个压缩器Compressor，实现了CompressionOutputStream类。提供数据压缩功能。
客户端压缩流程压缩器内部

Scott-Blog