mr分类输出（MultipleOutputs）

た入场券 2022-07-16 07:55 132阅读 0赞

原始数据：

[INFO]127.0.0.1 2016-04-14 09:06:33  vi 200
    [DEBUG]127.0.0.1 2016-04-13 09:06:40  rm 200
    [FETAL]127.0.0.1 2016-04-12 09:14:54  vi 300
    [WARNING]127.0.0.1 2016-04-15 09:24:58  vi 200
    [INFO]127.0.0.1 2016-04-16 09:25:43  vi 200
    [INFO]127.0.0.1 2016-04-17 09:27:02  mv 400
    [INFO]127.0.0.1 2016-04-18 09:06:33 vi 200
    [DEBUG]127.0.0.1 2016-04-19 09:06:40 rm 200
    [FETAL]127.0.0.1 2016-04-15 09:14:54 vi 300
    [INFO]127.0.0.1 2016-04-20 09:25:43 vi 200
    [INFO]127.0.0.1 2016-04-16 09:27:02 mv 400
    [INFO]127.0.0.1 2016-04-18 09:06:33 vi 200
    [DEBUG]127.0.0.1 2016-04-17 09:06:40 rm 200
    [FETAL]127.0.0.1 2016-04-19 09:14:54 vi 300
    [INFO]127.0.0.1 2016-04-16 09:25:43 vi 200
    [INFO]127.0.0.1 2016-04-19 09:27:02 mv 400
    [INFO]127.0.0.1 2016-04-17 09:06:33 vi 200

要求：按照日期分别存放不同的文件  
代码如下：

package mr.inputANDoutput;
    
    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.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.output.FileOutputFormat;
    import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
    
    
    /** * 根据不同的日期把文件分类保存 * @author 韩利鹏 * */
    public class OutputFormat { 
    
        /** * map方法 * @author 韩利鹏 * 输出的key是日期 * 输出的value是原始数据 */
        public static class OutMapper extends Mapper<LongWritable, Text, Text, Text>{ 
            Text text=new Text();//用来存放时间
            @Override
            protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
                    throws IOException, InterruptedException {
                String[] strs=value.toString().split(" ");
                text.set(strs[1]);
                context.write(text,value);
            }
        }
    
        /** * reduce 的输入的key value是map阶段的输出 * @author root * reduce 的输出的key 和 value时两个字符串 （key是空字符串） * 使用MultipleOutputs的时候注意需要重载setup 和 cleanup两个方法 在job里面也需要设置 */
        public static class OutReducer extends Reducer<Text, Text, Text, Text>{ 
            private MultipleOutputs<Text,Text> mo;
            @Override
            protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
                    throws IOException, InterruptedException {
                try{
                    String date = key.toString();
                    String year = date.substring(0,4);
                    String mouth = date.substring(5,7);
    
                    for(Text text:values){
                        mo.write("mulit", new Text(""), text, year+"/"+mouth+"/"+key);
                    }
                }catch(Exception e){
                    System.out.println("数据格式有误key值："+key.toString()+"value值为："+values.toString());
                }
            }
            @Override
            protected void setup(Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
                mo = new MultipleOutputs<Text,Text>(context);
                super.setup(context);
            }
    
            @Override
            protected void cleanup(Reducer<Text, Text, Text, Text>.Context context)
                    throws IOException, InterruptedException {
                mo.close();
                super.cleanup(context);
            }
        }
    
        /** * 启动mr的入口 * @param args * @throws IOException * @throws Exception */
        public static void main(String[] args) throws IOException, Exception{
            Configuration config=new Configuration();
            config.set("fs.defaultFS", "hdfs://192.168.10.97:9000/");
    
            Job job=Job.getInstance(config);
            //设置主类
            job.setJarByClass(OutputFormat.class);
            //设置map的类
            job.setMapperClass(OutMapper.class);
            //设置reduce的主类
            job.setReducerClass(OutReducer.class);
            //设置map的输出的key 和 value
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);
            //设置reduce的key 和 value 
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);
            //设置输入和输出路径
            FileInputFormat.addInputPath(job, new Path("/file/format"));
            FileOutputFormat.setOutputPath(job, new Path("/han/outputformat"));  
    
            MultipleOutputs.addNamedOutput(job, "mulit",org.apache.hadoop.mapreduce.lib.output.TextOutputFormat.class , Text.class, Text.class);
    
            //提交job作业
            System.exit(job.waitForCompletion(true)?0:1);
    
        }
    
    }