13.MapReduce框架原理

13.3 Shuffle机制

13.3.7 WritableComparable排序案例实操（区内排序）

13.3.7.1 需求

要求每个省份手机号输出的文件中按照总流量内部排序。

13.3.7.2 需求分析

基于前一个需求，增加自定义分区类，分区按照省份手机号设置。

13.3.7.3 案例实操

创建一个PartitionerAndWritableComparable的文件夹，将writableComparable2里面4个java代码同时复制到PartitionerAndWritableComparable里面

13.3.7.3.1 增加自定义分区类

package com.summer.mapreduce.partitionerAndWritableComparable;import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;/*** @author Redamancy* @create 2022-10-04 13:27*/
public class ProvincePartitioner2 extends Partitioner<FlowBean, Text> {@Overridepublic int getPartition(FlowBean flowBean, Text text, int i) {//获取手机号前三位prePhoneString phone = text.toString();String prePhone = phone.substring(0,3);//定义一个分区号变量partition, 根据prePhone 设置分区号int partition;if("136".equals(prePhone)){partition = 0;}else if("137".equals(prePhone)){partition = 1;}else if("138".equals(prePhone)){partition = 2;}else if("139".equals(prePhone)){partition = 3;}else {partition = 4;}//最后返回分区号partitionreturn partition;}
}

13.3.7.3.2在驱动类中添加分区类

在FlowDriver里面添加指定自定义分区器同时指定相应数量的ReduceTask

运行完成，结果和预想的一样，over！

13.3.8 Combiner合并

（1）Combiner是MR程序中Mapper和Reducer之外的一种组件。
（2）Combiner组件的父类就是Reducer。
（3）Combiner和Reducer的区别在于运行的位置
Combiner是在每一个MapTask所在的节点运行;
Reducer是接收全局所有Mapper的输出结果；
（4）Combiner的意义就是对每一个MapTask的输出进行局部汇总，以减小网络传输量。
（5）Combiner能够应用的前提是不能影响最终的业务逻辑，而且，Combiner的输出kv应该跟Reducer的输入kv类型要对应起来。

13.3.8.1 自定义 Combiner 实现步骤

13.3.8.1.1 自定义一个 Combiner 继承 Reducer，重写 Reduce 方法

package com.summer.mapreduce.combiner;import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;/*** @author Redamancy* @create 2022-10-04 17:21*/
public class WordCountCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {private IntWritable outV = new IntWritable();@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {int sum = 0;for (IntWritable value : values) {sum += value.get();}outV.set(sum);context.write(key, outV);}
}

13.3.8.1.2 在 Job 驱动类中设置

// 指定需要使用combiner，以及用哪个类作为combiner的逻辑
job.setCombinerClass(WordCountCombiner.class);

package com.summer.mapreduce.combiner;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.IOException;/*** @author Redamancy* @create 2022-08-22 17:23*/public class WordCountDriver {public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {//1 获取jobConfiguration conf = new Configuration();Job job = Job.getInstance(conf);//2 设置jar包路径job.setJarByClass(WordCountDriver.class);//3 关联mapper和reduccerjob.setMapperClass(WordCountMapper.class);job.setReducerClass(WordCountReducer.class);//4 设置map输出的kv类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);//5 设置最终输出的kv类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);// 指定需要使用combiner，以及用哪个类作为combiner的逻辑job.setCombinerClass(WordCountCombiner.class);//6 设置输入路径和输出路径FileInputFormat.setInputPaths(job, new Path("D:\\Acode\\Hadoop\\input\\inputhello"));//输出的路径为空，要是有该文件，则会报错FileOutputFormat.setOutputPath(job, new Path("D:\\Acode\\Hadoop\\output\\output1"));//7 提交jobboolean result = job.waitForCompletion(true);System.exit(result ? 0 : 1);}
}