Hadoop MultipleOutputs输出到多个文件中的实现方法

private MultipleOutputs mos;
public void setup(Context context) throws IOException,InterruptedException {
  mos = new MultipleOutputs(context);
}
public void cleanup(Context context) throws IOException,InterruptedException {
  mos.close();
}

1512,iphone5s,4英寸,指纹识别,A7处理器,64位,M7协处理器,低功耗

1512,iphone5,4英寸,A6处理器,IOS7

1512,iphone4s,3.5英寸,A5处理器,双核,经典

50019780,ipad,9.7英寸,retina屏幕,丰富的应用

50019780,yoga,联想,待机18小时,外形独特

50019780,nexus 7,华硕&google,7英寸

50019780,ipad mini 2,retina显示屏,苹果,7.9英寸

1101,macbook air,苹果超薄,OS X mavericks

1101,macbook pro,苹果,OS X lion

1101,thinkpad yoga,联想,windows 8,超级本

package cn.edu.bjut.multioutput;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MultiOutPutMapper extends Mapper {

  @Override
  protected void map(LongWritable key, Text value, Context context)
      throws IOException, InterruptedException {
    String line = value.toString().trim();
    if(null != line && 0 != line.length()) {
      String[] arr = line.split(",");
      context.write(new IntWritable(Integer.parseInt(arr[0])), value);
    }
  }

}

package cn.edu.bjut.multioutput;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

public class MultiOutPutReducer extends
    Reducer {

  private MultipleOutputs multipleOutputs = null;

  @Override
  protected void reduce(IntWritable key, Iterable values, Context context)
      throws IOException, InterruptedException {
    for(Text text : values) {
      multipleOutputs.write("KeySpilt", NullWritable.get(), text, key.toString()+"/");
      multipleOutputs.write("AllPart", NullWritable.get(), text);
    }
  }

  @Override
  protected void setup(Context context)
      throws IOException, InterruptedException {
    multipleOutputs = new MultipleOutputs(context);
  }

  @Override
  protected void cleanup(Context context)
      throws IOException, InterruptedException {
    if(null != multipleOutputs) {
      multipleOutputs.close();
      multipleOutputs = null;
    }
  }


}

package cn.edu.bjut.multioutput;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
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.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class MainJob {
  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = new Job(conf, "aaa");
    job.setJarByClass(MainJob.class);

    job.setMapperClass(MultiOutPutMapper.class);
    job.setMapOutputKeyClass(IntWritable.class);
    job.setMapOutputValueClass(Text.class);

    job.setReducerClass(MultiOutPutReducer.class);
    job.setOutputKeyClass(NullWritable.class);
    job.setOutputValueClass(Text.class);

    FileInputFormat.addInputPath(job, new Path(args[0]));

    MultipleOutputs.addNamedOutput(job, "KeySpilt", TextOutputFormat.class, NullWritable.class, Text.class);
    MultipleOutputs.addNamedOutput(job, "AllPart", TextOutputFormat.class, NullWritable.class, Text.class);

    Path outPath = new Path(args[1]);
    FileSystem fs = FileSystem.get(conf);
    if(fs.exists(outPath)) {
      fs.delete(outPath, true);
    }
    FileOutputFormat.setOutputPath(job, outPath);

    job.waitForCompletion(true);
  }
}