/**
 * MapReduce压缩支持的位置一共有如下几个位置：
 * 1. Map的输入文件可以支持压缩
 * 2. Map输出数据的时候也支持压缩
 * 3. Reduce输出数据的时候也支持压缩
 *
 * MapReduce程序只需要管压缩，不需要管解压缩。因为MR程序在读取压缩文件的时候会根据压缩文件类型自动解压缩
 * 压缩说白了就是把一个文件去创建一个输入IO流，然后创建一个压缩工具的输出IO流，将输入流的数据传递给输出IO流就可以实现压缩
 * @Author: ZYD
 * @Date: 2021/8/16 下午 17:26
 */
public class Demo {
    public static void main(String[] args) throws Exception {
        /**
         * 测试Hadoop的压缩机制----将一个文件压缩成为Hadoop支持的压缩包
         */
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://192.168.218.55:9000"), conf, "root");
        // 指定压缩文件
        String fileName = "/school/web.log";
        // 指定使用的压缩工具
//        String className = "org.apache.hadoop.io.compress.DefaultCodec";
//        String className = "org.apache.hadoop.io.compress.GzipCodec";
        String className = "org.apache.hadoop.io.compress.BZip2Codec";
        // 代表根据压缩工具的全限定类名创建一个压缩工具的编解码器
        CompressionCodec cc = (CompressionCodec) ReflectionUtils.newInstance(Class.forName(className), conf);
        String outPath = fileName + ".bzip2";
        // 压缩就是将文件重新以编码器的IO流输出即可
        // 创建fileName的输入流
        FSDataInputStream inputStream = fs.open(new Path(fileName));

        // 解压缩
        // 创建一个普通的获取文件的输出流
        FSDataOutputStream fsDataOutputStream = fs.create(new Path(outPath));
        // 根据普通的输出流创建一个 压缩输出流
        CompressionOutputStream outputStream = cc.createOutputStream(fsDataOutputStream);

        IOUtils.copyBytes(inputStream, outputStream, 1024*1024, false);

        outputStream.close();
        inputStream.close();
    }
}

public class OpenDemo {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://192.168.218.55:9000"), conf, "root");
        /**
         * 解压缩压缩包
         */
        // 1. 校验压缩包支不支持解压缩
        CompressionCodecFactory factory = new CompressionCodecFactory(conf);
        // 2. 判断当前文件用的什么压缩编码器，如果没有对应解压缩编码器，则返回null
        Path p = new Path("/school/web.log.deflate");
        CompressionCodec codec = factory.getCodec(p);
        if (codec == null) {
            System.out.println("当前文件不支持解压缩");
        } else {
            // 创建文件输出流
            CompressionInputStream inputStream = codec.createInputStream(fs.open(p));
            // 创建解压缩输入流
            FSDataOutputStream fsDataOutputStream = fs.create(new Path("/school/web-deflate.log"));
            IOUtils.copyBytes(inputStream, fsDataOutputStream, 1024*1024, false);
            fsDataOutputStream.close();
            inputStream.close();
        }
    }
}

// 开启map输出
conf.setBoolean("mapreduce.map.output.compress", true);
// 一旦map端开启输出，我们就要指定map压缩采用什么压缩机制
conf.setClass("mapreduce.map.output.compress.codec", DefaultCodec.class, CompressionCodec.class);

// 给reduce端开启压缩
FileOutputFormat.setCompressOutput(job, true);
// reduce端输出使用上面类型的压缩格式
FileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class);