MapReduce的常见输入格式之NlineInputFormat

小灰灰 2022-11-14 13:28 150阅读 0赞

有两个文件log ,log副本

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNDUwMzUyNg_size_16_color_FFFFFF_t_70][]

*  切片策略： 读取配置文件中的参数`mapreduce.input.lineinputformat.linespermap`，默认为1，以文件为单位，切片每此参数行作为1片！
 *  既然有参数，那就可以修改，设置为每N行切为一片：
 *      Configuration conf = new Configuration();
        conf.set("mapreduce.input.lineinputformat.linespermap", "2")

`RecordReader`：`LineRecordReader`，一次处理一行，将一行内容的偏移量作为key，一行内容作为value

所以上面两个文件总共八行，若一行切一片，则有八片；两行切一片，则有四片。

用户编写的程序分成三个部分：Mapper、Reducer 和 Driver。

**1.创建WCMapper.java**

public class WCMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
    
        private Text out_key=new Text();
        private IntWritable out_value=new IntWritable(1);
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    
            System.out.println("keyin:"+key+"----keyout:"+value);
    
            String[] words = value.toString().split("\t");
    
            for (String word : words) {
                out_key.set(word);
                //写出数据(单词,1)
                context.write(out_key, out_value);
            }
    
        }
    }

2.创建**WCReducer.java**

public class WCReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
    
        private IntWritable out_value=new IntWritable();
    
        // reduce一次处理一组数据，key相同的视为一组
        @Override
        protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException{
    
            int sum=0;
    
            for (IntWritable intWritable : values) {
                sum+=intWritable.get(); 
            }
    
            out_value.set(sum);
    
            //将累加的值写出
            context.write(key, out_value);
    
        }
    }

3.创建**WCDriver.java**

conf.set("mapreduce.input.lineinputformat.linespermap", "2");//设置为每两行切一片

// 声明使用NLineInputFormat  
   job.setInputFormatClass(NLineInputFormat.class);

public class WCDriver {
    
        public static void main(String[] args) throws Exception {
    
            Path inputPath=new Path("d:/mrinput");
            Path outputPath=new Path("d:/mroutput");
    
            //作为整个Job的配置
            Configuration conf = new Configuration();
    
            conf.set("mapreduce.input.lineinputformat.linespermap", "2");//设置为每两行切一片
    
            //保证输出目录不存在
            FileSystem fs=FileSystem.get(conf);
    
            if (fs.exists(outputPath)) {
                fs.delete(outputPath, true);
            }
    
            // ①创建Job
            Job job = Job.getInstance(conf);
    
            job.setJarByClass(WCDriver.class);
    
            // ②设置Job
            // 设置Job运行的Mapper，Reducer类型，Mapper,Reducer输出的key-value类型
            job.setMapperClass(WCMapper.class);
            job.setReducerClass(WCReducer.class);
    
            // Job需要根据Mapper和Reducer输出的Key-value类型准备序列化器，通过序列化器对输出的key-value进行序列化和反序列化
            // 如果Mapper和Reducer输出的Key-value类型一致，直接设置Job最终的输出类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            // 声明使用NLineInputFormat
            job.setInputFormatClass(NLineInputFormat.class);
    
            // 设置输入目录和输出目录
            FileInputFormat.setInputPaths(job, inputPath);
            FileOutputFormat.setOutputPath(job, outputPath);
    
            // ③运行Job
            job.waitForCompletion(true);
    
    
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNDUwMzUyNg_size_16_color_FFFFFF_t_70]: /images/20221022/22eed5c8f70c48fcadbf0c26f44a53fe.png