title: mapreduce多目錄輸出筆記
date: 2016/11/26 22:23:21
tags: MapReduce
categories: 大數(shù)據(jù)

使用MultipleOutputs實(shí)現(xiàn)多目錄/文件輸出

org.apache.hadoop.mapreduce.lib.output.MultipleOutputs

在map或者reduce類中加入如下方法

    private MultipleOutputs<Text, NullWritable> mos;

        @Override
        protected void setup(Reducer<Text, NullWritable, Text, NullWritable>.Context context)
                throws IOException, InterruptedException {
            // TODO Auto-generated method stub
            super.setup(context);
            mos = new MultipleOutputs<Text, NullWritable>(context);// 初始化mos
        }

        @Override
        protected void cleanup(Reducer<Text, NullWritable, Text, NullWritable>.Context context)
                throws IOException, InterruptedException {
            // TODO Auto-generated method stub
            super.cleanup(context);
            mos.close();
        }

在需要輸出數(shù)據(jù)的地方卖词，可以使用定義好的 mos 進(jìn)行輸出

mos.write("outputName", key, value);
mos.write("outputName", key, value, "filePrefix"); 
mos.write("outputName", key, value, "path/filePrefix");//到文件夾

在Job Driver 時(shí)定義一些 Named Output

MultipleOutputs.addNamedOutput(job, "outputXName",
    XXXOutputFormat.class, OutputXKey.class, OutputXValue.class);
MultipleOutputs.addNamedOutput(job, "outputYName",
    YYYOutputFormat.class, OutputYKey.class, OutputYValue.class);

取消類似part-r-00000的空文件
LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class)
例子

package com.hdu.recommend.mr;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.yarn.conf.YarnConfiguration;

 * @author Skye
 *
 */
public class DataCleanIconAndWeb {
    public static class QLMapper extends
            Mapper<LongWritable, Text, Text, NullWritable> {

        private String webGame = "網(wǎng)頁游戲";

        Text outputValue = new Text();
        // 設(shè)置多文件輸出
        private MultipleOutputs<Text,NullWritable> mos;
        @Override
        protected void setup(Mapper<LongWritable, Text, Text, NullWritable>.Context context)
                throws IOException, InterruptedException {
            // TODO Auto-generated method stub
            super.setup(context);
            mos = new MultipleOutputs<Text, NullWritable>(context);// 初始化mos
        }
        @Override
        protected void cleanup(Mapper<LongWritable, Text, Text, NullWritable>.Context context)
                throws IOException, InterruptedException {
            // TODO Auto-generated method stub
            super.cleanup(context);
            mos.close();
        }
        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            // 接收數(shù)據(jù)v1
            String line = value.toString();
            // 切分?jǐn)?shù)據(jù)
            String[] words = line.split("?");
            // String[] words = line.split("\t");
            boolean isWeb = false;
            boolean flag = true;
            
            //一系列處理代碼
            //***
            //***
            //***
            String action = words[1] + "\t" + words[0] + "\t" + words[2]
                        + "\t" + words[3] + "\t" + words[5];

            outputValue.set(action);
            mos.write("iconRecord", outputValue, NullWritable.get(),"iconRecord/icon");
            
            
    
            String action = words[1] + "\t" + words[0] + "\t"
                            + words[2] + "\t" + words[3] + "\t" + words[4]
                            + "\t" + words[5];

            outputValue.set(action);
            mos.write( "webRecord",outputValue, NullWritable.get(),"webRecord/web");
                

            
        }

    }

    

    public static void run(String originalDataPath, String dataCleanOutputFile)
            throws Exception {

        // 構(gòu)建Job對(duì)象
        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);

        // 注意：main方法所在的類
        job.setJarByClass(DataCleanIconAndWeb.class);
        job.getConfiguration().setBoolean("mapreduce.output.fileoutputformat.compress", false);
        job.getConfiguration().setStrings(
                "mapreduce.reduce.shuffle.input.buffer.percent", "0.1");
        job.getConfiguration().setBoolean("mapreduce.output.fileoutputformat.compress", false);
        job.setNumReduceTasks(3);

        // 設(shè)置Mapper相關(guān)屬性
        job.setMapperClass(QLMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        
        FileInputFormat.setInputPaths(job, new Path(originalDataPath));

        

        // 設(shè)置Reducer相關(guān)屬性
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        FileOutputFormat.setOutputPath(job, new Path(dataCleanOutputFile));
        
        MultipleOutputs.addNamedOutput(job, "iconRecord",
                TextOutputFormat.class, Text.class, NullWritable.class);
        MultipleOutputs.addNamedOutput(job, "webRecord",
                TextOutputFormat.class, Text.class, NullWritable.class);
        
        // 文件格式
        job.setInputFormatClass(TextInputFormat.class);
        //取消part-r-00000新式文件輸出
        LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);
        
        
        //job.setOutputFormatClass(TextOutputFormat.class);
        // 提交任務(wù)
        job.waitForCompletion(true);

        long endTime = System.currentTimeMillis();

    }
 
}

參考
http://gnailuy.com/dataplatform/2015/11/22/common-techniques-for-mapreduce/
http://blog.csdn.net/zgc625238677/article/details/51524786
https://www.iteblog.com/archives/848