import org.apache.hadoop.io.Text;

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.ContextFactory;
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.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MapFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import net.sourceforge.pinyin4j.PinyinHelper;
import net.sourceforge.pinyin4j.format.HanyuPinyinCaseType;
import net.sourceforge.pinyin4j.format.HanyuPinyinOutputFormat;
import net.sourceforge.pinyin4j.format.HanyuPinyinToneType;
import net.sourceforge.pinyin4j.format.HanyuPinyinVCharType;
import net.sourceforge.pinyin4j.format.exception.BadHanyuPinyinOutputFormatCombination;

public class MultiFileOut extends Configured implements Tool {
    private final static String[] Shengarry = { "北京", "天津", "山西", "内蒙古", "辽宁",
            "吉林", "黑龙江", "上海", "江苏", "浙江", "安徽", "福建", "江西", "山东", "河南", "湖北",
            "湖南", "广东", "广西", "海南", "重庆", "四川", "贵州", "云南", "西藏", "陕西省", "甘肃",
            "青海", "宁夏", "新疆", "河北" };
    private final static String[] sexary = { "M", "F" };

    public static String getPinYin(String src) {
        char[] srcary = null;
        srcary = src.toCharArray();
        String[] strtmp = new String[srcary.length];

        // 设置汉字拼音输出的格式
        HanyuPinyinOutputFormat formatstr = new HanyuPinyinOutputFormat();
        formatstr.setCaseType(HanyuPinyinCaseType.LOWERCASE);
        formatstr.setToneType(HanyuPinyinToneType.WITHOUT_TONE);
        formatstr.setVCharType(HanyuPinyinVCharType.WITH_V);
        String resultstr = "";
        int t0 = srcary.length;
        try {
            for (int i = 0; i < t0; i++) {
                // 判断能否为汉字字符
                if (Character.toString(srcary[i]).matches("[\\u4E00-\\u9FA5]+")) {
                    strtmp = PinyinHelper.toHanyuPinyinStringArray(srcary[i],
                            formatstr);// 将汉字的几种全拼都存到t2数组中
                    resultstr += strtmp[0];// +" ";// 取出该汉字全拼的第一种读音并连接到字符串t4后
                } else {
                    // 如果不是汉字字符，间接取出字符并连接到字符串t4后
                    resultstr += Character.toString(srcary[i]);
                }
            }
        } catch (BadHanyuPinyinOutputFormatCombination e) {
            e.printStackTrace();
        }
        return resultstr;
    }

    private static class ProvinceMapper extends
            Mapper<Object, Text, Text, Text> {
        @Override
        protected void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            String str = value.toString();
            String outkey = "";
            Boolean isfind = false;
            if (str.indexOf("name") >= 0)
                return;

            String[] strArray = str.split(",");
            if (strArray.length != 33)
                return;

            String sex = strArray[5];
            String addr = strArray[7];
            for (int i = 0; i < Shengarry.length; i++) {
                for (int j = 0; j < sexary.length; j++) {
                    int index = addr.indexOf(Shengarry[i]);
                    if ((index >= 0) && (index <= 3)
                            && (sex.indexOf(sexary[j]) >= 0)) {
                        isfind = true;
                        outkey = getPinYin(Shengarry[i]) + sexary[j];
                        break;
                    }
                }

                if (isfind)
                    break;
            }

            if (isfind) {
                context.write(new Text(outkey), value);
            } else {
                System.out.println("Error Data" + value.toString());
            }
        }
    }

    private static class ProvinceReducer extends
            Reducer<Text, Text, NullWritable, Text> {
        private MultipleOutputs mos = null;

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

        @Override
        protected void cleanup(Context context) throws IOException,
                InterruptedException {
            mos.close();
        }

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            Text value = http://www.mamicode.com/new Text("");
            String valuetmp = "";

            for (Text va : values) {
                value.set(va.toString());
                
                try {
                    mos.write(key.toString(), NullWritable.get(), value);
                } catch (Exception e) {
                    //System.out.println("Exception" + key);
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new Configuration(), new MultiFileOut(), args);
    }

    @Override
    public int run(String[] args) throws Exception {
        int result = 0;
        Configuration conf = new Configuration();
        String[] argArray = new GenericOptionsParser(conf, args)
                .getRemainingArgs();
        if (argArray.length != 2) {
            System.err.println("Usage: MultiFileOut <in> <out>");
            System.exit(1);
        }

        Job job = new Job(conf, "MultiFileOut");
        job.setJarByClass(MultiFileOut.class);
        job.setMapperClass(ProvinceMapper.class);
        job.setReducerClass(ProvinceReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        // job.setOutputFormatClass(WordCountOutputFormat.class);
        FileInputFormat.addInputPath(job, new Path(argArray[0]));
        FileOutputFormat.setOutputPath(job, new Path(argArray[1]));
        
        for (int i = 0; i < Shengarry.length; i++) {
            for (int j = 0; j < sexary.length; j++) {
                MultipleOutputs.addNamedOutput(job, getPinYin(Shengarry[i])
                        + sexary[j], TextOutputFormat.class, Text.class,
                        Text.class);
            }
        }
        
        try {
            result = job.waitForCompletion(true) ? 0 : 1;
        } catch (ClassNotFoundException | InterruptedException e) {
            e.printStackTrace();
        }

        return result;
    }
}