hadoop压缩框架

public static void compress(String method) throws…… {  
  File fileIn = new File("README.txt");  
 
 //输入流  
  InputStream in =  new FileInputStream(fileIn);  
 
 Class<?> codecClass = Class.forName(method);  
 
 Configuration conf = new Configuration();  
  //通过名称找对应的编码/解码器  
  CompressionCodec codec = (CompressionCodec)  
        ReflectionUtils.newInstance(codecClass, conf);  
 
 File fileOut = new File("README.txt"+codec.getDefaultExtension());  
  fileOut.delete();  
  //文件输出流  
  OutputStream out =  new FileOutputStream(fileOut);  
 
 //通过编码/解码器创建对应的输出流  
  CompressionOutputStream cout =  
     codec.createOutputStream(out);  
 
 //压缩  
  IOUtils.copyBytes(in, cout, 4096, false);  
 
 in.close();  
 cout.close();  
}  

public static void decompress(File file) throws IOException {  
  Configuration conf = new Configuration();  
  CompressionCodecFactory factory = new CompressionCodecFactory(conf);  
 
 //通过文件扩展名获得相应的编码/解码器  
  CompressionCodec codec = factory.getCodec(new Path(file.getName()));  
 
 if( codec == null ) {  
     System.out.println("Cannot find codec for file "+file);  
     return;  
  }  
 
 File fileOut = new File(file.getName()+".txt");  
 
 //通过编码/解码器创建对应的输入流  
  InputStream in = codec.createInputStream( new FileInputStream(file) );  
  ……  
}  

public interface CompressionCodec {  
 
  //在底层输出流out的基础上创建对应压缩算法的压缩流CompressionOutputStream对象  
  CompressionOutputStream createOutputStream(OutputStream out)……  
 
  //使用压缩器compressor，在底层输出流out的基础上创建对应的压缩流  
  CompressionOutputStream createOutputStream(OutputStream out,  
                                             Compressor compressor) ……  
  ……  
  //创建压缩算法对应的压缩器  
  Compressor createCompressor();  
 
 //在底层输入流in的基础上创建对应压缩算法的解压缩流CompressionInputStream对象  
  CompressionInputStream createInputStream(InputStream in) ……  
  ……  
 
 //获得压缩算法对应的文件扩展名  
  String getDefaultExtension();  
}  

{  
  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  
} 

public class CompressionCodecFactory {  
  ……  
  //该有序映射保存了逆转文件后缀（包括后缀前的“.”）到CompressionCodec的映射  
  //通过逆转文件后缀，我们可以找到最长匹配后缀  
  private SortedMap<String, CompressionCodec> codecs = null;  
  ……  
  public CompressionCodec getCodec(Path file) {  
        CompressionCodec result = null;  
     if (codecs != null) {  
        String filefilename = file.getName();  
        //逆转字符串  
        String reversedFilename = new 
           StringBuffer(filename).reverse().toString();  
        SortedMap<String, CompressionCodec> subMap =  
           codecs.headMap(reversedFilename);  
        if (!subMap.isEmpty()) {  
           String potentialSuffix = subMap.lastKey();  
           if (reversedFilename.startsWith(potentialSuffix)) {  
              result = codecs.get(potentialSuffix);  
           }  
        }  
     }  
     return result;  
  }  
} 

public interface Compressor {
  /**
   * 输入要压缩的数据. 
   * 根据#needsInput() 返回的值，判断是否执行 
   * 如果返回为true，表示需要更多的数据.
   * 
   * @param b 输入数据
   * @param off 偏移的起始位置
   * @param len 长度
   */
  public void setInput(byte[] b, int off, int len);
  
  /**
   * 当输入数据缓存为空的时候返回 true，并且
   * #setInput() 应该被调用用来提供输入数据. 
   */
  public boolean needsInput();
  
  /**
   * 设置当前路径为压缩.
   *
   * @param b Dictionary data bytes
   * @param off Start offset
   * @param len Length
   */
  public void setDictionary(byte[] b, int off, int len);

  /**
   * 返回输入的还未被压缩的字节数.
   */
  public long getBytesRead();

  /**
   * 返回输出的已经被压缩的字节数.
   */
  public long getBytesWritten();

  /**
   * 当调用的时候，表示当前输入缓存的内容的压缩操作应该结束
   */
  public void finish();
  
  /**
   * 当压缩器中的最末端的输出数据到达则返回true.
   */
  public boolean finished();
  
  /**
   * 向缓存中填充压缩的数据. 返回实际的压缩的字节数. 返回0表明 needsInput()被调用
   * 来判断是否需要更多的输入数据.
   * 
   * @param b Buffer for the compressed data
   * @param off Start offset of the data
   * @param len Size of the buffer
   * @return The actual number of bytes of compressed data.
   */
  public int compress(byte[] b, int off, int len) throws IOException;
  
  /**
   * 重置 compressor， 预备一些新的输入数据.
   */
  public void reset();
  
  /**
   * 关闭 compressor并丢弃为处理的输入.
   */
  public void end(); 
}

public static void compressor() throws ClassNotFoundException, IOException  
{  
  //读入被压缩的内容  
  File fileIn = new File("README.txt");  
  InputStream in =  new FileInputStream(fileIn);  
  int datalength=in.available();  
  byte[] inbuf = new byte[datalength];  
  in.read(inbuf, 0, datalength);  
  in.close();  
 
 //长度受限制的输出缓冲区，用于说明finished()方法  
  byte[] outbuf = new byte[compressorOutputBufferSize];  
 
 Compressor compressor=new BuiltInZlibDeflater();//构造压缩器  
 
 int step=100;//一些计数器  
  int inputPos=0;  
  int putcount=0;  
  int getcount=0;  
  int compressedlen=0;  
  while(inputPos < datalength) {  
     //进行多次setInput()  
     int len=(datalength-inputPos>=step)? step:datalength-inputPos;  
     compressor.setInput(inbuf, inputPos, len );  
     putcount++;  
 
    while (!compressor.needsInput()) {  
        compressedlen=compressor.compress(outbuf, 0, ……);  
        if(compressedlen>0) {  
           getcount++;  //能读到数据  
        }  
     }  // end of while (!compressor.needsInput())  
     inputPos+=step;  
  }  
 
 compressor.finish();  
 
 while(!compressor.finished()) {  //压缩器中有数据  
     getcount++;  
     compressor.compress(outbuf, 0, compressorOutputBufferSize);  
  }  
 
 System.out.println("Compress "+compressor.getBytesRead()  //输出信息  
                     +" bytes into "+compressor.getBytesWritten());  
  System.out.println("put "+putcount+" times and get "+getcount+" times");  
 
 compressor.end();//停止  
}  

public abstract class CompressionOutputStream extends OutputStream {  
  //输出压缩结果的流  
  protected final OutputStream out;  
 
 //构造函数  
  protected CompressionOutputStream(OutputStream out) {  
     this.out = out;  
  }  
 
 public void close() throws IOException {  
     finish();  
     out.close();  
  }  
 
 public void flush() throws IOException {  
     out.flush();  
  }  
  public abstract void write(byte[] b, int off, int len) throws IOException;  
 
 public abstract void finish() throws IOException;  
 
 public abstract void resetState() throws IOException;  
}  

public class CompressorStream extends CompressionOutputStream {  
  protected Compressor compressor;  
  protected byte[] buffer;  
  protected boolean closed = false;  
 
 //构造函数  
  public CompressorStream(OutputStream out,  
           Compressor compressor, int bufferSize) {  
     super(out);  
     ……//参数检查，略  
     this.compressor = compressor;  
     buffer = new byte[bufferSize];  
  }  
  ……  
  public void write(byte[] b, int off, int len) throws IOException {  
     //参数检查，略  
     ……  
     compressor.setInput(b, off, len);  
     while (!compressor.needsInput()) {  
        compress();  
     }  
  }  
 
 protected void compress() throws IOException {  
     int len = compressor.compress(buffer, 0, buffer.length);  
     if (len > 0) {  
        out.write(buffer, 0, len);  
     }  
  }  
  //结束输入  
  public void finish() throws IOException {  
     if (!compressor.finished()) {  
        compressor.finish();  
        while (!compressor.finished()) {  
           compress();  
        }  
     }  
  }  
  ……  
  //关闭流  
  public void close() throws IOException {  
     if (!closed) {  
        finish();//结束压缩  
        out.close();//关闭底层流  
        closed = true;  
     }  
  }  
  ……  
}  

public class SnappyCompressor implements Compressor {  
  ……  
  private native static void initIDs();  
  private native int compressBytesDirect();  
}