多文件输出

package country;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
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 Multiples {

    public static void main(String[] args) throws Exception {
        System.setProperty("hadoop.home.dir", "F:\\JAVA\\hadoop-2.2.0");
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(Multiples.class);

        job.setMapperClass(SBCMapper.class);

        /**
         * 使用KeyValueTextInputFormat作为输入类型
         */
        job.setInputFormatClass(KeyValueTextInputFormat.class);

        /**
         * 指定key和Value的分隔符【默认也是\t】
         */
        //conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, "\t"); 

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setNumReduceTasks(0);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);        

        /**
         * 注意在初始化时需要设置输出文件的名 
         * 另外名称，不支持中文名，仅支持英文字符 
         */
        // 1.job    2.namedOutput    3.OutputFormat    4.keyClass    5.valueClass
        MultipleOutputs.addNamedOutput(job,"china",TextOutputFormat.class,Text.class,Text.class);
        MultipleOutputs.addNamedOutput(job,"USA",TextOutputFormat.class,Text.class,Text.class);
        MultipleOutputs.addNamedOutput(job,"CPerson",TextOutputFormat.class,Text.class,Text.class);

        KeyValueTextInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.waitForCompletion(true);
    }

    //Map阶段
    public static class SBCMapper extends Mapper<Text, Text, Text, Text>{

        //声明多集合输出类型MultipleOutputs
        MultipleOutputs<Text, Text> mos=null;

        @Override
        //此方法在map任务执行前【只会执行一次】
        protected void setup(Context context)
                throws IOException, InterruptedException {

            //初始化，只会执行一次
            mos=new MultipleOutputs<Text, Text>(context);
        }        
        @Override
        protected void map(Text key, Text value, Context context)
                throws IOException, InterruptedException {

            String line=new String(key.getBytes(),0,key.getLength(),"utf-8");

            if(line.equalsIgnoreCase("中国")){
                mos.write("china", key, value);
            }else if (line.equalsIgnoreCase("美国")) {
                mos.write("USA", key, value);
            }else if (line.equalsIgnoreCase("中国人")) {
                mos.write("CPerson", key, value);
            }
        }
        @Override
        protected void cleanup(Context context)
                throws IOException, InterruptedException {

            //关闭输出流
            mos.close();
        }        
    }    
}