hadoop案例:NLine (分行统计)

逃离我推掉我的手 2022-11-21 14:54 54阅读 0赞

### 文章目录 ###

*   *  需求
     *  输入数据
     *  期望输出数据
     *  需求分析
     *  编写代码
     *   *  Mapper类
         *  Reducer类
         *  Driver类
         *  执行结果

## 需求 ##

对每个单词进行个数统计，要求根据每个输入文件的行数来规定输出多少个切片。此案例要求每三行放入一个切片中。

## 输入数据 ##

Nu.txt

banzhang ni hao
    xihuan hadoop banzhang
    banzhang ni hao
    xihuan hadoop banzhang
    banzhang ni hao
    xihuan hadoop banzhang
    banzhang ni hao
    xihuan hadoop banzhang
    banzhang ni hao
    xihuan hadoop banzhang banzhang ni hao
    xihuan hadoop banzhang

## 期望输出数据 ##

Number of splits:4

## 需求分析 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center]

## 编写代码 ##

### Mapper类 ###

package com.mr.nline;
    
    import com.mr.wordcount.WordCountMapper;
    import com.mr.wordcount.WordCountReducer;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    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.input.NLineInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    
    
    public class NlineWordCountDriver { 
    
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
    
            // 0 指定路径
            args = new String[]{ "E:/Hadoop/src/main/resources/input/nline", "E:/Hadoop/src/main/resources/ouput/nline"};
    
    // 1 获取配置信息configuration以及封装任务job
            Configuration configuration = new Configuration();
            Job job = Job.getInstance(configuration);
    
    // 设置每个切片的行数
            NLineInputFormat.setNumLinesPerSplit(job, 3);
    // 将默认的TextInputFormat替换成NLineInputFormat
            job.setInputFormatClass(NLineInputFormat.class);
    
    // 2 设置Driver加载路径 setJarByClass
            job.setJarByClass(NlineWordCountDriver.class);
    
    // 3 设置map和reduce类 setMaper setReducer
            job.setMapperClass(WordCountMapper.class);
            job.setReducerClass(WordCountReducer.class);
    
    // 4 设置map输出 setmapoutputkey setmapoutputvalue
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
    
    // 5 设置最终输出kv类型 (reducer的输出kv类型) setoutoutkey setoutputvalue
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
    // 6 设置本地的输入和输出路径 fileinputformat.setinputpath
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
    // 7 提交
            boolean completion = job.waitForCompletion(true);
            System.exit(completion ? 0 : 1);
        }
    }

### Reducer类 ###

package com.mr.nline;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    
    public class NlineWordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> { 
    
        // 0. 将创建对象的操作提取成变量，防止在 map 方法重复创建
        private Text text = new Text();
        private IntWritable i = new IntWritable(1);
    
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
            // 1. 将 Hadoop 内置的text 数据类型转换为string类型
            // 方便操作
            String string = value.toString();
            System.out.println("key:" + key);
    
            // 2. 对字符串进行切分
            String[] split = string.split(" ");
    
            // 3. 对字符串数组遍历，将单词映射成 （单词，1）
            for (String s : split) { 
                text.set(s);
                context.write(text, i);
            }
    
    
        }
    }

### Driver类 ###

package com.mr.nline;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    
    public class NlineWordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> { 
    
        private IntWritable total = new IntWritable();
    
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { 
    
            // 定义一个 sum，用来对每个键值对的 值 做 累加操作
            int sum = 0;
    
            for (IntWritable value : values) { 
                int i = value.get();
                sum += i;
            }
            total.set(sum);
            // 最后写出到文件
            context.write(key, total);
    
        }
    }

### 执行结果 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/60fdb0cd4a9a45eab6b92fc1c70becbb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/f0cb963b2a5b41c594f8a0385db3d685.png