MapReduce词频统计【自定义复杂类型、自定义Partitioner、NullWritable使用介绍】

缺乏、安全感 2022-02-23 11:14 336阅读 0赞

## 一、MapReduce1.0运行模型 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70][]

## 二、MapReduce编程模型之执行步骤  ##

1、准备map处理的输入数据

2、交给**Mapper进行处理**

3、Shuffle【规则可以自己控制】

4、**Reduce处理**\[合并、归并\]

5、输出

## MapReduce处理流程 ##

InputFormat读数据，通过Split将数据切片成InputSplit，通过RecordReader读取记录，再交给map处理，处理后输出一个临时的<k,v>键值对，再将结果交给shuffle处理，最终在reduce中将最后处理后的<k,v>键值对结果通过OutputFormat重新写回到HDFS中。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 1][]

## 三、词频统计原理图： ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 2][]

**Combiner能减少网络IO、提升作业的性能**

**Combiner的局限性**：**求平均数：**总数 / 个数   **对于含有除法的操作，需要慎重，有可能结果会不正确**

## 四、词频统计具体代码实现\[读写在HDFS和本地完成\] ##

0、pom.xml

<?xml version="1.0" encoding="UTF-8"?>
 <project xmlns="http://maven.apache.org/POM/4.0.0"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>
 
 <groupId>cn.itcats</groupId>
 <artifactId>hadoop-mapreduce</artifactId>
 <version>1.0-SNAPSHOT</version>
 
 <dependencies>
 <dependency>
 <groupId>org.apache.hadoop</groupId>
 <artifactId>hadoop-client</artifactId>
 <version>2.6.0</version>
 </dependency>
 
 <dependency>
 <groupId>junit</groupId>
 <artifactId>junit</artifactId>
 <version>4.12</version>
 </dependency>
 </dependencies>
 
 </project>

## ##

1、准备一个自定义的Mapper类

package cn.itcats.hadoop.mapreduce.wordcount;
 
 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;
 
 /**
 * Mapper类4个泛型的含义
 * KEYIN: Map任务读数据的key类型,offset,是每行数据起始位置的偏移量,Long(Java)
 * VALUEIN:Map任务读数据的value类型,其实就是一行行的字符串,String
 *
 * 如文本中的数据为 :
 * hello world welcome
 * hello welcome
 *
 * KEYOUT: map方法自定义实现输出的key的类型,String
 * VALUEOUT:map方法自定义实现输出的value类型,Integer
 *
 * 词频统计: 相同单词的次数 (word,1)
 * Long,String,String,Integer是Java里面的数据类型
 * 因为涉及网络传输,需要序列化与反序列化
 * 使用Hadoop提供的自定义类型:
 * Long => LongWritable String => Text Integer => IntWritable
 */
 
 //词频统计
 public class WordCountMapper extends Mapper<LongWritable,Text,Text,IntWritable> {
 @Override
 protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
 //把value对应的行数据按照指定的分隔符拆开
 String[] words = value.toString().split(",");
 for(String word : words){
 //(hello,1) (word,1)
 //转成小写,忽略大小写
 context.write(new Text(word.toLowerCase()) , new IntWritable(1));
 }
 }
 }

2、准备一个自定义的Reducer类

package cn.itcats.hadoop.mapreduce.wordcount;
 
 import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Reducer;
 
 import java.io.IOException;
 import java.util.Iterator;
 
 public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
 /**
 * (hello,1) (world,1)
 * (hello,1) (world,1)
 * (hello,1) (world,1)
 * (welcome,1)
 * 
 * key 为 word values含义:
 * map的输出到reduce端,是按照相同的key分发到一个reduce上去执行
 * reduce1 : (hello,1) (hello,1) (hello,1) => (hello,[1,1,1])
 * reduce2 : (world,1) (world,1) (world,1) => (world,[1,1,1])
 * reduce3 : (welcome,1) => (welcome,[1])
 *
 * Reducer和Mapper中使用到了什么设计模式? 模板模式
 */
 
 protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
 int count = 0;
 Iterator<IntWritable> iterator = values.iterator();
 while (iterator.hasNext()){
 IntWritable value = iterator.next();
 count += value.get();
 }
 context.write(key, new IntWritable(count));
 
 }
 }

### 读写在HDFS完成 ###

3、准备一个任务驱动类

package cn.itcats.hadoop.mapreduce.wordcount;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 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.output.FileOutputFormat;
 
 import java.net.URI;
 
 
 /**
 * 使用MapReduce统计HDFS上文件对应的词频
 * 
 * Driver: 配置Mapper,Reducer的相关属性
 * 
 * 提交到HDFS运行
 *
 * 含有Combiner操作
 */
 public class WordCountCombinerApp {
 public static void main(String[] args) throws Exception {
 
 System.setProperty("HADOOP_USER_NAME", "root");
 
 Configuration configuration = new Configuration();
 configuration.set("fs.defaultFS", "hdfs://hdp-01:9000");
 
 // 创建一个Job
 Job job = Job.getInstance(configuration);
 
 // 设置Job对应的主类、Mapper、Reducer类
 job.setJarByClass(WordCountCombinerApp.class);
 job.setMapperClass(WordCountMapper.class);
 job.setReducerClass(WordCountReducer.class);
 
 //添加Combiner的设置
 job.setCombinerClass(WordCountReducer.class);
 
 //设置Job对应的参数: Mapper输出key和value的类型
 job.setMapOutputKeyClass(Text.class);
 job.setMapOutputValueClass(IntWritable.class);
 
 //设置Job对应的参数: Reducer输出key和value的类型
 job.setOutputKeyClass(Text.class);
 job.setOutputValueClass(IntWritable.class);
 
 //如果文件已经存在则先删除,否则会报错org.apache.hadoop.mapred.FileAlreadyExistsException
 //获取FileSystem对象进行exists/delete操作
 FileSystem fileSystem = FileSystem.get(new URI("hdfs://hdp-01:9000"), configuration, "root");
 Path outputPath = new Path("/wordcount/output");
 if (fileSystem.exists(outputPath)) {
 fileSystem.delete(outputPath,true);
 }
 
 //设置job作业输入和输出的路径
 FileInputFormat.setInputPaths(job, new Path("/wordcount/input"));
 FileOutputFormat.setOutputPath(job, outputPath);
 
 //提交job
 boolean resullt = job.waitForCompletion(true);
 
 System.exit(resullt ? 0 : 1);
 }
 }

4、将文件上传到HDFS的/wordcount/input中(没有创建提前创建好该文件夹)

1.txt

hello,world,welcome,hello,welcome,Welcome,Hello,haha,Haha,hAha

### 读写在本地中完成 ###

将上述的第四步改为：

在工程下创建一个input文件夹

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 3][]

修改上述第三步的代码，只new Configuration()即可，最后把输入路径改成input、输出路径改成output即可

package cn.itcats.hadoop.mapreduce.wordcount;
    
    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.output.FileOutputFormat;
    
    import java.io.File;
    
    
    /**
     * 使用MapReduce统计HDFS上文件对应的词频
     *
     * Driver: 配置Mapper,Reducer的相关属性
     *
     * 提交到本地运行运行(使用本地文件进行统计,统计结果输出到本地路径)
     */
    public class WordCountLocalApp {
        public static void main(String[] args) throws Exception {
    
            Configuration configuration = new Configuration();
    
            // 创建一个Job
            Job job = Job.getInstance(configuration);
    
            // 设置Job对应的主类、Mapper、Reducer类
            job.setJarByClass(WordCountLocalApp.class);
            job.setMapperClass(WordCountMapper.class);
            job.setReducerClass(WordCountReducer.class);
    
            //设置Job对应的参数: Mapper输出key和value的类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
    
            //设置Job对应的参数: Reducer输出key和value的类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            //设置job作业输入和输出的路径
            FileInputFormat.setInputPaths(job,new Path("input"));
            FileOutputFormat.setOutputPath(job,new Path("output"));
    
            //提交job
            boolean resullt = job.waitForCompletion(true);
    
            System.exit(resullt ? 0 : 1);
        }
    }

### 运行测试 ###

运行WordCountLocalApp

得到最终结果输出：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 4][]

输出结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 5][]

# #

## 五、关于自定义复杂类型的介绍 ##

package cn.itcats.hadoop.mapreduce.access;
    
    /*
     * 自定义复杂的数据类型
     * 对此Hadoop有一些规范
     * 1、需要实现Writable
     * 2、需要实现write和readFields这两个方法
     * 3、需要定义默认的构造方法
     */
    
    import lombok.Data;
    import org.apache.hadoop.io.Writable;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    @Data
    public class Access implements Writable {
        private String phone;
        private long up;
        private long down;
        private long sum;
    
        public void write(DataOutput dataOutput) throws IOException {
            dataOutput.writeUTF(phone); //字符串
            dataOutput.writeLong(up);   //Long
            dataOutput.writeLong(down);
            dataOutput.writeLong(sum);
        }
    
        public void readFields(DataInput dataInput) throws IOException {
            //规范: 严格按照上面写的顺序
            this.phone = dataInput.readUTF();
            this.up = dataInput.readLong();
            this.down = dataInput.readLong();
            this.sum = dataInput.readLong();
        }
    
        public Access() {
        }
    }

## 六、NullWritable介绍 ##

若Mapper或Reducer中某个输入或输出不想输出显示，则使用NullWritable替换我们常用的类型(如Text、LongWritable)等

Plus:**NullWritable.get()，返回NullWritable类型**

# #

## 七、自定义Partitioner介绍 ##

默认情况下Map的输出需要做shuffle操作，将key根据一定的算法分发到Reduce上执行【如相同的key，或者具有相似特征的key】，我们也可以自定义分区写数据

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 6][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 7][]

**numReduceTasks：你的作业所指定的reducer的个数，决定了reduce作业输出文件的个数**

**HashPartitioner是MapReduce默认的分区规则**

//泛型对应map的输出(KEYOUT和VALUEOUT)
 public class AccessPartitioner extends Partitioner<Text,Access>{
 
 //根据手机号开头数字,分派到不同的分区
 public int getPartition(Text phone, Access access, int numPartitions) {
 //总共定义了4种分区规则,后面设置分区数也要填4
 if(phone.toString().startsWith("13")){
 return 0;
 }else if(phone.toString().startsWith("18")){
 return 1;
 }else if(phone.toString().startsWith("15")){
 return 2;
 }else{
 return 3;
 }
 }
 }

在job任务驱动类中加上一行代码：

//设置自定义分区规则
    job.setPartitionerClass(AccessPartitioner.class);
    //设置reduce个数
    job.setNumReduceTasks(4);

观察output文件输出：【总共4个文件】

part-r-00000
    part-r-00001
    part-r-00002
    part-r-00003

**总结：Partitioner决定maptask输出的数据交由哪个reducetask处理**

**默认实现：分发的key的hash值与reduce task个数取模**

### **八、程序修改为在Yarn上运行** ###

第一步

那么outputPath、和inputPath则不能写死，修改为args\[0\]、args\[1\]

package cn.itcats.hadoop.mapreduce.wordcount;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 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.output.FileOutputFormat;
 
 import java.net.URI;
 
 
 /**
 * 使用MapReduce统计HDFS上文件对应的词频
 * 
 * Driver: 配置Mapper,Reducer的相关属性
 * 
 * 提交到HDFS运行
 *
 * 含有Combiner操作
 */
 public class WordCountYarnApp {
 public static void main(String[] args) throws Exception {
 
 System.setProperty("HADOOP_USER_NAME", "root");
 
 Configuration configuration = new Configuration();
 configuration.set("fs.defaultFS", "hdfs://hdp-01:9000");
 
 // 创建一个Job
 Job job = Job.getInstance(configuration);
 
 // 设置Job对应的主类、Mapper、Reducer类
 job.setJarByClass(WordCountYarnApp.class);
 job.setMapperClass(WordCountMapper.class);
 job.setReducerClass(WordCountReducer.class);
 
 //添加Combiner的设置
 job.setCombinerClass(WordCountReducer.class);
 
 //设置Job对应的参数: Mapper输出key和value的类型
 job.setMapOutputKeyClass(Text.class);
 job.setMapOutputValueClass(IntWritable.class);
 
 //设置Job对应的参数: Reducer输出key和value的类型
 job.setOutputKeyClass(Text.class);
 job.setOutputValueClass(IntWritable.class);
 
 //如果文件已经存在则先删除,否则会报错org.apache.hadoop.mapred.FileAlreadyExistsException
 //获取FileSystem对象进行exists/delete操作
 FileSystem fileSystem = FileSystem.get(new URI("hdfs://hdp-01:9000"), configuration, "root");
 Path outputPath = new Path(args[0]);
 if (fileSystem.exists(outputPath)) {
 fileSystem.delete(outputPath,true);
 }
 
 //设置job作业输入和输出的路径
 FileInputFormat.setInputPaths(job, new Path(args[1]));
 FileOutputFormat.setOutputPath(job, outputPath);
 
 //提交job
 boolean result = job.waitForCompletion(true);
 
 System.exit(result ? 0 : 1);
 }
 }

第二步

**进入工程，执行maven打包命令**

mvn clean package -DskipTests

打包完成后jar包在当前项目的 target/文件夹内

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 8][]

在hadoop机器上执行

hadoop jar hadoop-mapreduce-1.0-SNAPSHOT.jar 完整类名 args[0] arg[1]
    
    //其中上面的args[0]、args[1]都是我们修改源码后的参数，对应输入路径和输出路径,填上执行即可

总结：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 9][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70]: /images/20220223/215a121f26a245a2b7752e05ac086341.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 1]: /images/20220223/c01cd2cb5d62476c8c44a78e2e9ffaaf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 2]: /images/20220223/2b9f5544879c41478a7ed665497cf4b8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 3]: /images/20220223/6b15ceb210e041bdae1680463d348eec.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 4]: /images/20220223/a220a0cbf04446e09d3498ddc85e1bcd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 5]: /images/20220223/9ecf41c2d7084a3cb4b3b3b7faa7e22b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 6]: /images/20220223/01b183856a5f4277aeb0c8f44fcee653.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 7]: /images/20220223/2af5cc5ccdea47f0aef6d4cd1da17faa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 8]: /images/20220223/1ebf99c401fc4935a0860da990fa83a7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 9]: /images/20220223/42f5ac8c660a4a46be1d90d754e88d4f.png