hadoop3.x MapReduce 代码设置map、reduce压缩

作者：马育民 • • 阅读：10185

说明

map输出结果压缩 和 reduce输出结果压缩，是完全独立的，互相不影响

map输出结果压缩后，reduce读取其结果时，会自动解压缩

注意：

由于使用的是 snappy 方式压缩，在windows上不支持，所以不要在windows上测试

设置 map 输出结果压缩

修改 driver 类

方式一

// 开启map输出压缩
config.setBoolean("mapreduce.map.output.compress", true);
// 设置map输出压缩方式
config.set("mapreduce.map.output.compress.codec", "org.apache.hadoop.io.compress.SnappyCodec");

方式二

// 开启map输出压缩
config.setBoolean("mapreduce.map.output.compress", true);
// 设置map输出压缩方式
config.setClass("mapreduce.map.output.compress.codec", SnappyCodec.class,CompressionCodec.class);

设置 reduce 结果压缩

修改 driver 类

方式一

// 开启reduce输出压缩
config.setBoolean("mapreduce.output.fileoutputformat.compress", true);
// 设置reduce输出压缩方式
config.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.SnappyCodec");

方式二

// 设置reduce端输出压缩开启
FileOutputFormat.setCompressOutput(job, true);
// 设置压缩的方式
FileOutputFormat.setOutputCompressorClass(job, SnappyCodec.class);

例子

只需要修改 driver 类，如下：

package compress;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Main2 extends Configured implements Tool{
    public static void main(String[] args) throws Exception {

        Configuration config = new Configuration();
        // 开启map输出压缩
        config.setBoolean("mapreduce.map.output.compress", true);
         // 设置map输出压缩方式
//        config.setClass("mapreduce.map.output.compress.codec", SnappyCodec.class,CompressionCodec.class);    
        config.set("mapreduce.map.output.compress.codec", "org.apache.hadoop.io.compress.SnappyCodec");    
        // 开启reduce输出压缩
        config.setBoolean("mapreduce.output.fileoutputformat.compress", true);
        // 设置reduce输出压缩方式
          config.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.SnappyCodec");
        //通过ToolRunner类 执行 覆盖的run()方法
        int res=ToolRunner.run(config, new Main2(), args);
        //退出客户端
        System.exit(res);
    }


    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(getConf(),"压缩");
        //设置jar加载路径，不加此代码有时可能会报错
        job.setJarByClass(Main2.class);
        //设置map和reduce类
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);
        //设置map输出 key/value 类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Cdz.class);
        //设置最终输出 key/value 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Cdz.class);
        //根据输入参数 设置输入和输出路径

        String os = System.getProperty("os.name").toLowerCase();

        String inPath;
        String outPath;
        if(os.startsWith("win")){
            inPath="D:\\bigdata\\chongdianzhuang\\chongdianzhuang2.txt";
            outPath="D:\\bigdata\\chongdianzhuang\\res_compress";
        }else{
            inPath=args[0];
            outPath=args[1];
        }
        FileInputFormat.setInputPaths(job, new Path(inPath));
        FileOutputFormat.setOutputPath(job, new Path(outPath));

        // 设置reduce端输出压缩开启
//         FileOutputFormat.setCompressOutput(job, true);

         // 设置压缩的方式
//         FileOutputFormat.setOutputCompressorClass(job, SnappyCodec.class);

        //提交
//        job.submit();
        boolean result = job.waitForCompletion(true);
        return result?1:0;
    }
}

测试

注意：由于使用的是 snappy 方式压缩，在windows上不支持，所以不要在windows上测试

在 yarn 上测试：

hadoop jar /program/myjar/chongdianzhuang-0.0.1-SNAPSHOT.jar  compress.Main2  /chongdianzhan/chongdianzhan.txt /chongdianzhan/compress_res9

控制台结果

可以看到 map输出结果 压缩了

注意： reduce 读取 map压缩结果时，会自动解压缩

查看 9870

可以看到结果文件的后缀是 .snappy ，说明 reduce 压缩配置成功

原文出处：/show_1IX2EAStNV1O.html

吉ICP备15003043号