Hadoop压缩之CompressionCodecFactory

  1 package cn.roboson.codec;
  2 
  3 import java.io.IOException;
  4 
  5 import org.apache.hadoop.conf.Configuration;
  6 import org.apache.hadoop.fs.FSDataInputStream;
  7 import org.apache.hadoop.fs.FSDataOutputStream;
  8 import org.apache.hadoop.fs.FileStatus;
  9 import org.apache.hadoop.fs.FileSystem;
 10 import org.apache.hadoop.fs.Path;
 11 import org.apache.hadoop.io.IOUtils;
 12 import org.apache.hadoop.io.compress.CompressionCodec;
 13 import org.apache.hadoop.io.compress.CompressionCodecFactory;
 14 import org.apache.hadoop.io.compress.CompressionInputStream;
 15 
 16 /*
 17  * 通过CompressionCodeFactory推断CompressionCodec
 18  * 1.先从本地上传一个.gz后缀的文件到Hadoop
 19  * 2.通过文件后缀推断出所用的压缩算法
 20  * 3.解压上传的压缩文件到统一个目录下
 21  */
 22 public class StreamCompressor02 {
 23     
 24     public static void main(String[] args) {
 25         
 26         Configuration conf = new Configuration();
 27         conf.addResource("core-site.xml");
 28         
 29         try {
 30             FileSystem fs = FileSystem.get(conf);
 31             
 32             //本地文件
 33             String localsrc="/home/roboson/桌面/README.txt.gz";
 34             Path localPath= new Path(localsrc);
 35             
 36             //目的处路径
 37             String hadoopdsc="/roboson/README.txt.gz";
 38             Path hadoopPath = new Path(hadoopdsc);
 39             
 40             //复制前/roboson目录下的文件列表
 41             FileStatus[] files = fs.listStatus(new Path("/roboson/"));
 42             System.out.println("复制前：");
 43             for (FileStatus fileStatus : files) {
 44                 System.out.println(fileStatus.getPath());
 45             }
 46             
 47             //复制本地文件到Hadoop文件系统中
 48             fs.copyFromLocalFile(localPath,hadoopPath);
 49             
 50             //复制后/roboson目录下的文件列表
 51             files = fs.listStatus(new Path("/roboson/"));
 52             System.out.println("复制后：");
 53             for (FileStatus fileStatus : files) {
 54                 System.out.println(fileStatus.getPath());
 55             }
 56             
 57             //获得一个CompressionCodecFactory实例来推断哪种压缩算法
 58             CompressionCodecFactory facotry = new CompressionCodecFactory(conf);
 59             
 60             //通过CompressionCodecFactory推断出一个压缩类，用于解压
 61             CompressionCodec codec =facotry.getCodec(hadoopPath);
 62             if(codec==null){
 63                 System.out.println("没有找到该类压缩");
 64                 System.exit(1);
 65             }
 66             
 67             /*
 68              * 1.CompressionCodecFactory的removeSuffix（）用来返回一个文件名，这个文件名==压缩文件的后缀名去掉
 69              * 如README.txt.gz调用removeSuffix（）方法后，返回的是README.txt
 70              * 
 71              * 2.CompressionCodec的getDefaultExtension()方法返回的是一个压缩算法的压缩扩展名，如gzip的是.gz
 72              */
 73             String uncodecUrl=facotry.removeSuffix(hadoopdsc, codec.getDefaultExtension());
 74             System.out.println("压缩算法的生成文件的扩展名："+codec.getDefaultExtension());
 75             System.out.println("解压后生成的文件名："+uncodecUrl);
 76             
 77             //在Hadoop中创建解压后的文件
 78             FSDataOutputStream out = fs.create(new Path(uncodecUrl));
 79             
 80             //创建输入数据流,并用CompressionCodec的createInputStream()方法，将输入数据流中读取的数据解压
 81             FSDataInputStream in = fs.open(new Path(hadoopdsc));
 82             CompressionInputStream codecIn = codec.createInputStream(in);
 83             
 84             //将输入数据流写入到 输出数据流
 85             IOUtils.copyBytes(codecIn, out, conf,true);
 86             
 87             //解压后/roboson目录下的文件列表
 88             files = fs.listStatus(new Path("/roboson/"));
 89             System.out.println("解压后");
 90             for (FileStatus fileStatus : files) {
 91                 System.out.println(fileStatus.getPath());
 92             }
 93             
 94             //查看解压后的内容
 95             System.out.println("解压后的内容：");
 96             in=fs.open(new Path(uncodecUrl));
 97             IOUtils.copyBytes(in,System.out, conf,true);
 98         } catch (IOException e) {
 99             // TODO Auto-generated catch block
100             e.printStackTrace();
101         }
102     }
103 }