Hadoop-MapReduce-WritableComparable排序、全排序、区内排序案例实操-连载中

红太狼 2023-03-01 10:49 38阅读 0赞

### 文章目录 ###

*   *   *  1 WritableComparable排序
         *  2 WritableComparable排序案例实操（全排序）
         *  3 WritableComparable排序案例实操（区内排序）

### 1 WritableComparable排序 ###

**排序概述**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 1]  
**排序分类**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 2]  
**自定义排序WritableComparable原理分析**

bean对象做为key传输，需要实现WritableComparable接口重写compareTo方法，就可以实现排序。

@Override
    public int compareTo(FlowBean bean) {
        
    	int result;	
    	// 按照总流量大小，倒序排列
    	if (this.sumFlow > bean.getSumFlow()) {
        
    		result = -1;
    	}else if (this.sumFlow < bean.getSumFlow()) {
        
    		result = 1;
    	}else {
        
    		result = 0;
    	}
    	return result;
    }

### 2 WritableComparable排序案例实操（全排序） ###

**1）需求**

根据案例2.3序列化案例产生的结果再次对总流量进行倒序排序。

（1）输入数据

原始数据：第一次处理后的数据**part-r-00000**

（2）期望输出数据

> 13509468723 7335 110349 117684
> 
> 13736230513 2481 24681 27162
> 
> 13956435636 132 1512 1644
> 
> 13846544121 264 0 264
> 
> 。。。 。。

**2）需求分析**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 3]  
**3）代码实现**

（1）FlowBean对象在在需求1基础上增加了比较功能

import org.apache.hadoop.io.WritableComparable;
    
    public class FlowBean implements WritableComparable<FlowBean> {
        
    	private long upFlow;
    	private long downFlow;
    	private long sumFlow;
        
    	// 反序列化时，需要反射调用空参构造函数，所以必须有
    	public FlowBean() {
        
    	}
    	// 序列化方法
    	@Override
    	public void write(DataOutput out) throws IOException {
        
    		out.writeLong(upFlow);
    		out.writeLong(downFlow);
    		out.writeLong(sumFlow);
    	}
    
    	// 反序列化方法 注意反序列化的顺序和序列化的顺序完全一致 
    	@Override
    	public void readFields(DataInput in) throws IOException {
        
    		upFlow = in.readLong();
    		downFlow = in.readLong();
    		sumFlow = in.readLong();
    	}
    
    	@Override
    	public String toString() {
        
    		return upFlow + "\t" + downFlow + "\t" + sumFlow;
    	}
    
    	@Override
    	public int compareTo(FlowBean bean) {
        	
    		int result;	
    		// 按照总流量大小，倒序排列
    		if (sumFlow > bean.getSumFlow()) result = -1;
    		else if (sumFlow < bean.getSumFlow()) result = 1;
    		else result = 0;
    		return result;
    	}
    }

（2）编写Mapper类

public class FlowCountSortMapper extends Mapper<LongWritable, Text, FlowBean, Text>{
        
    
    	FlowBean bean = new FlowBean();
    	Text v = new Text();
        
    	@Override
    	protected void map(LongWritable key, Text value, Context context)	throws IOException, InterruptedException {
        
    		// 1 获取一行
    		String line = value.toString();
    		// 2 截取
    		String[] fields = line.split("\t");
    		// 3 封装对象
    		String phoneNbr = fields[0];
    		long upFlow = Long.parseLong(fields[1]);
    		long downFlow = Long.parseLong(fields[2]);
    		
    		bean.set(upFlow, downFlow);
    		v.set(phoneNbr);
    		// 4 输出
    		context.write(bean, v);
    	}
    }

（3）编写Reducer类

public class FlowCountSortReducer extends Reducer<FlowBean, Text, Text, FlowBean>{
        
    
    	@Override
    	protected void reduce(FlowBean key, Iterable<Text> values, Context context)	throws IOException, InterruptedException {
        	
    		// 循环输出，避免总流量相同情况
    		for (Text text : values) {
        
    			context.write(text, key);
    		}
    	}
    }

（4）编写Driver类

public class FlowCountSortDriver {
        
    
    	public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException {
        
    		// 输入输出路径需要根据自己电脑上实际的输入输出路径设置
    		args = new String[]{
        "e:/output","e:/output"};
    		// 1 获取配置信息，或者job对象实例
    		Configuration configuration = new Configuration();
    		Job job = Job.getInstance(configuration);
    		// 2 指定本程序的jar包所在的本地路径
    		job.setJarByClass(FlowCountSortDriver.class);
    		// 3 指定本业务job要使用的mapper/Reducer业务类
    		job.setMapperClass(FlowCountSortMapper.class);
    		job.setReducerClass(FlowCountSortReducer.class);
    		// 4 指定mapper输出数据的kv类型
    		job.setMapOutputKeyClass(FlowBean.class);
    		job.setMapOutputValueClass(Text.class);
    		// 5 指定最终输出的数据的kv类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(FlowBean.class);
    		// 6 指定job的输入原始文件所在目录
    		FileInputFormat.setInputPaths(job, new Path(args[0]));
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    		// 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
    		boolean result = job.waitForCompletion(true);
    		System.exit(result ? 0 : 1);
    	}
    }

### 3 WritableComparable排序案例实操（区内排序） ###

**1）需求**

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

**2）需求分析**

基于前一个需求，增加自定义分区类，分区按照省份手机号设置。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 4]  
**3）案例实操**

（1）增加自定义分区类

public class ProvincePartitioner extends Partitioner<FlowBean, Text> {
        
    
    	@Override
    	public int getPartition(FlowBean key, Text value, int numPartitions) {
        	
    		// 1 获取手机号码前三位
    		String preNum = value.toString().substring(0, 3);	
    		int partition = 4;	
    		// 2 根据手机号归属地设置分区
    		if ("136".equals(preNum)) {
        
    			partition = 0;
    		}else if ("137".equals(preNum)) {
        
    			partition = 1;
    		}else if ("138".equals(preNum)) {
        
    			partition = 2;
    		}else if ("139".equals(preNum)) {
        
    			partition = 3;
    		}
    		return partition;
    	}
    }

（2）在驱动类中添加分区类

// 加载自定义分区类
    job.setPartitionerClass(ProvincePartitioner.class);
    
    // 设置Reducetask个数
    job.setNumReduceTasks(5);

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70]: /images/20230208/21d3586eb6e94523b3e7870737faf658.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 1]: /images/20230208/231d837835424459adcc4576425ff2a1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 2]: /images/20230208/ae6fea5dcaa14d97991d5b024d67b79c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 3]: /images/20230208/046e49b61281492f96d5e9213534d8ca.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200724125221533.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1,size_16,color_FFFFFF,t_70