//开启map端的输入压缩
conf.setBoolean("mapreduce.map.output.compress",true);//设置压缩方法
//默认
conf.setClass("mapreduce.map.output.compress.codec",DefaultCodeC.class,CompressionCodec.class);
//Bzip2
conf.setClass("mapreduce.mapt.output.compress.codec",Bzip2Codec.class,CompressionCode.class);
//LZO
conf.setClass("mapreduce.map.output.compress.codec".LZOCodec.class,CompressionCodec.class);

//开启reduce端的输出压缩
FileOutputFormat.setCompressOutput(job,true);
//设置压缩方法
//默认
FileOutputFormat.setOutputCompressorClass(job,DefaultCodec.class);
//Bzip2
FileOutputFormat.setOutputCompressorClass(job,Bzip2Codec.class);
//Gzip
FileOutputFormat.setOutputCompressorClass(job,GzipCodec.class);

/*** @author: PrincessHug* @date: 2019/4/8, 9:49* @Blog: https://www.cnblogs.com/HelloBigTable/*/
public class TestCompress {public static void main(String[] args) throws IOException, ClassNotFoundException {Compress("G:\\weblog.log","org.apache.hadoop.io.compress.BZip2Codec");}//自定义压缩方法private static void Compress(String fileName,String method) throws IOException, ClassNotFoundException {//获取输入流FileInputStream fis = new FileInputStream(new File(fileName));//通过反射获取压缩方法并初始化Class cName = Class.forName(method);CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(cName, new Configuration());//定义输出流FileOutputStream fos = new FileOutputStream(new File(fileName + codec.getDefaultExtension()));//创建压缩输出流CompressionOutputStream cos = codec.createOutputStream(fos);//流的拷贝IOUtils.copyBytes(fis,cos,2*1024*1024,false);//关闭资源fis.close();cos.close();fos.close();}
}