Hadoop MR 之(二) 传递自定义数据类型 & 排序

Dear 丶 2022-02-23 02:40 341阅读 0赞

### 前言 ###

前章, 我们介绍了简单的WordCount代码的基本实现([Hadoop MR 之(一) 编写自己的WordCount][Hadoop MR _ _WordCount]). 本章, 我们将介绍, 如何在`Map/Reduce`操作内传递自己的数据类型, 并根据部分关键值对进行排序操作.

本文相关代码, 可在我的Github项目 [https://github.com/SeanYanxml/bigdata/][https_github.com_SeanYanxml_bigdata] 目录下可以找到. PS: (如果觉得项目不错, 可以给我一个Star.)

--------------------

### 基础知识 ###

我们都知道, 要想数据进行网络传输, 必须先将其进行序列化操作. Java自带的`Serializable`接口就是其一. Hadoop自身也提供了一些列的序列化接口, 可供我们选择.

*  IntWritabel
 *  LongWritable
 *  DoubleWritable
 *  NullWritable
 *  Text
 *  Writable (通用类型)

--------------------

### 业务背景 ###

本章我们将要进行处理的数据为`flowdata`.

1836294111 10 20
    1836294112 1  2
    1836294113 2 3
    1836294114 3 4
    1836294115 2 1
    1836294116 1 2

其列值含义分别为: <手机号,上行流量,下行流量>. 本文, 我们的任务是计算出总流量, 并将其以总流量的大小进行排序.

##### 前置操作 #####

我们将这个文件拷贝到`/flowcount/input`目录下:

hadoop fs -mkdir /flowcount/input
    hadoop fs -put flowdata /flowcount/input

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

--------------------

### 基本操作 ###

*  导入相关Jar包 <pom.xml>文件略.(上章有提, 不再重述)
 *  编写业务类 FlowBean  
    通过`Writable`接口, 即可使其在`Map/Reduce`的操作内部进行传输.
    
     *  空构造函数, 供反射调用, 一定要?️;
     *  `write`/`readFields`方法是序列化和反序列化方法. 序列化方法的顺序要与反序列化一致. 读取`String`类型的数据, 需要使用`readUTF()`方法.
     *  `toString()`方法是为了输出调用. 即`output`.当然, 也可以自己编写`OutputFormat`.

public class FlowBean implements Writable{
    	private String phoneNumber;
    	private long upFlow;
    	private long downFlow;
    	public FlowBean(){}
    	public FlowBean(String phoneNumber, long upFlow, long downFlow){
    		this.phoneNumber = phoneNumber;
    		this.upFlow = upFlow;
    		this.downFlow = downFlow;
    	}
    	public FlowBean(long upFlow, long downFlow){
    		this.upFlow = upFlow;
    		this.downFlow = downFlow;
    	}
    	public String getPhoneNumber() {
    		return phoneNumber;
    	}
    	public void setPhoneNumber(String phoneNumber) {
    		this.phoneNumber = phoneNumber;
    	}
    	public long getUpFlow() {
    		return upFlow;
    	}
    	public void setUpFlow(long upFlow) {
    		this.upFlow = upFlow;
    	}
    	public long getDownFlow() {
    		return downFlow;
    	}
    	public void setDownFlow(long downFlow) {
    		this.downFlow = downFlow;
    	}
    	
    	/**
    	 * 序列化方法.
    	 * 
    	 * */
    	public void write(DataOutput out) throws IOException {
    		out.writeLong(upFlow);
    		out.writeLong(downFlow);
    	}
    	
    	/**
    	 * 反序列化方法.
    	 * 
    	 * */
    	public void readFields(DataInput in) throws IOException {
    		upFlow = in.readLong();
    		downFlow = in.readLong();
    	}
    	
    	public String toString(){
    		return "upFlow: "+upFlow+" downFlow: "+downFlow;
    	}
    }

*  编写`Mapper`& `Reducer` & `Job Driver`  
    这3点基本与`WordCount`例子一致, 我们将其快速写出来. 比较值得注意的是`Mapper<LongWritable, Text, Text, FlowBean>`其中的参数类型分别对应`<KEYIN,VALUEIN,KEYOUT,VALUEOUT>`即(`<传入KEY类型, 传入VALUE类型, 传出KEY类型, 传出VALUE类型>`).`Reducer`后的参数类型也是一样.

public class FlowCountDriver {
    	static class FlowCountMapper extends Mapper<LongWritable, Text, Text, FlowBean>{
    		protected void map(LongWritable key, Text value,Mapper<LongWritable, Text, Text, FlowBean>.Context context)throws IOException, InterruptedException {
    			// 将一行内容转换为string
    			String line = value.toString();
    			String []fields = line.split(" ");
    			//取出手机号
    			String phoneNumber = fields[0];
    			long upFlow = Long.parseLong(fields[1]);
    			long downFlow = Long.parseLong(fields[2]);
    			
    			context.write(new Text(phoneNumber), new FlowBean(phoneNumber,upFlow,downFlow));
    		}
    	}
    	
    	// <18362894222,1,2>...
    	static class FlowCountReducer extends Reducer<Text, FlowBean, Text, FlowBean>{
    
    		@Override
    		protected void reduce(Text key, Iterable<FlowBean> it,Reducer<Text, FlowBean, Text, FlowBean>.Context context)throws IOException, InterruptedException {
    			
    			String phoneNumber = key.toString();
    			long sum_upFlow = 0;
    			long sum_downFlow = 0;
    			
    			// 遍历所有的bean, 将其中的上行和下行流量进行累加
    			for(FlowBean bean:it){
    				sum_upFlow+=bean.getUpFlow();
    				sum_downFlow+=bean.getDownFlow();
    			}
    			
    			FlowBean resultBean = new FlowBean(phoneNumber,sum_upFlow, sum_downFlow);
    			context.write(key, resultBean);
    		}
    
    		@Override
    		protected void cleanup(Reducer<Text, FlowBean, Text, FlowBean>.Context context)throws IOException, InterruptedException {
    			super.cleanup(context);
    			// 可以在此处做排序操作.
    		}
    	}
    	
    	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    		Configuration conf =  new Configuration();
    		conf.set("mapreduce.framework.name", "yarn");
    		conf.set("yarn.resourcemanager.hostname", "localhost");
    		
    		Job job =  Job.getInstance(conf);
    		
    //		job.setJar("/");
    		// 指定本程序jar包所在地址
    		job.setJarByClass(FlowCountDriver.class);
    		
    		//指定本业务job需要使用的mapper业务类
    		job.setMapperClass(FlowCountMapper.class);
    		job.setReducerClass(FlowCountReducer.class);
    		
    		// 指定mapper输出数据的kv类型
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(FlowBean.class);
    		
    		// 指定最终输出数据的kv类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputKeyClass(FlowBean.class);
    		
    		// 设置匹配器
    		job.setPartitionerClass(ProvincePartitioner.class);
    //		job.setInputFormatClass(cls);
    		// 当数目小于分片数目时候 会报错. 多余时,则有的分片内为空.
    		job.setNumReduceTasks(5);
    		
    //		// 指定job的输入文件所在目录
    //		FileInputFormat.setInputPaths(job, new Path(args[0]));
    //		// 指定job的输出结果
    //        FileOutputFormat.setOutputPath(job,new Path(args[1]));
    		
    		// 指定job的输入文件所在目录
    		FileInputFormat.setInputPaths(job, new Path("/flowcount/input"));
    		// 指定job的输出结果
    		FileOutputFormat.setOutputPath(job, new Path("/flowcount/output"));
    		
    		// 将job中配置的相关参数, 以及job所用的java类所在的jar包,提交给yarn执行
    //		job.submit();
    		
    		// 
    		boolean flag = job.waitForCompletion(true);
    		System.exit(flag?0:1);
    	}
    }

--------------------

### 排序 ###

OK. 经过上述的操作, 我们可以通过`Map/Reduce`操作计算出总的流量, 并将其输出到目录内.

##### 排序1 使用`TreeSort`进行排序 #####

我们通过查看`Mapper`类的源码可以发现:

public void run(Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>.Context context) throws IOException, InterruptedException {
    		this.setup(context);
    		try {
    			while (context.nextKeyValue()) {
    				this.map(context.getCurrentKey(), context.getCurrentValue(), context);
    			}
    		} finally {
    			this.cleanup(context);
    		}
    	}

`Mapper`的`run()`方法相当于先通过`setup()`进行初始化操作, 随后循环调用`map()`方法, 最后`cleanup()`方法介绍线程. 我们排序的操作可以写在`cleanup()`中. 在Mapper处理业务逻辑后, 传入Ruduce之前进行排序. 带排序带Reducer程序如下所示:

// <18362894222,1,2>...
    	static class FlowCountReducer extends Reducer<Text, FlowBean, Text, FlowBean>{
    
    		Map <Text,FlowBean> sortMap = new HashMap();
    		@Override
    		protected void reduce(Text key, Iterable<FlowBean> it,Reducer<Text, FlowBean, Text, FlowBean>.Context context)throws IOException, InterruptedException {
    			
    			String phoneNumber = key.toString();
    			long sum_upFlow = 0;
    			long sum_downFlow = 0;
    			
    			// 遍历所有的bean, 将其中的上行和下行流量进行累加
    			for(FlowBean bean:it){
    				sum_upFlow+=bean.getUpFlow();
    				sum_downFlow+=bean.getDownFlow();
    			}
    			
    			FlowBean resultBean = new FlowBean(phoneNumber,sum_upFlow, sum_downFlow);
    			sortMap.put(key, resultBean);
    //			context.write(key, resultBean);
    		}
    
    		@Override
    		protected void cleanup(Reducer<Text, FlowBean, Text, FlowBean>.Context context)throws IOException, InterruptedException {
    //			context.
    			super.cleanup(context);
    			// 可以在此处做排序操作.
    			// 将Map转换为List用于排序操作
    		    List<Entry<Text, FlowBean>> list=new LinkedList<Map.Entry<Text,FlowBean>>(sortMap.entrySet());
    			Collections.sort(list,new Comparator<Map.Entry<Text,FlowBean>>() {
    
    				@Override
    				public int compare(Map.Entry<Text,FlowBean> o1, Map.Entry<Text,FlowBean> o2) {
    					String key1 = o1.toString();
    					String key2 = o2.toString();
    					return key1.compareTo(key2);
    				}
    				
    			});
    			// 循环遍历排序后的, 将其放入context内部
    			for(Entry<Text, FlowBean> entry:list){
    				context.write(entry.getKey(), entry.getValue());
    			}
    			
    		}
    	}

##### 排序2 使用`Map/Reduce`机制`KEY`进行排序 #####

Hadoop在传递参数时候, 默认会对`KEYOUT`进行排序. 利用这种特性, 我们可以将`Bean`放在`KEYOUT`位置进行传递. 相关代码如下所示:

*  重新`FlowBean`, 继承`WritableComparable`. 重新`compareTo()`方法.

public class FlowBean implements WritableComparable<FlowBean>{
    	private String phoneNumber;
    	private long upFlow;
    	private long downFlow;
    	public FlowBean(){}
    	public FlowBean(String phoneNumber, long upFlow, long downFlow){
    		this.phoneNumber = phoneNumber;
    		this.upFlow = upFlow;
    		this.downFlow = downFlow;
    	}
    	public FlowBean(long upFlow, long downFlow){
    		this.upFlow = upFlow;
    		this.downFlow = downFlow;
    	}
    	public String getPhoneNumber() {
    		return phoneNumber;
    	}
    	public void setPhoneNumber(String phoneNumber) {
    		this.phoneNumber = phoneNumber;
    	}
    	public long getUpFlow() {
    		return upFlow;
    	}
    	public void setUpFlow(long upFlow) {
    		this.upFlow = upFlow;
    	}
    	public long getDownFlow() {
    		return downFlow;
    	}
    	public void setDownFlow(long downFlow) {
    		this.downFlow = downFlow;
    	}
    	
    	/**
    	 * 序列化方法.
    	 * 
    	 * */
    	public void write(DataOutput out) throws IOException {
    		out.writeLong(upFlow);
    		out.writeLong(downFlow);
    	}
    	
    	/**
    	 * 反序列化方法.
    	 * 
    	 * */
    	public void readFields(DataInput in) throws IOException {
    		upFlow = in.readLong();
    		downFlow = in.readLong();
    	}
    	
    	public String toString(){
    		return "upFlow: "+upFlow+" downFlow: "+downFlow;
    	}
    	@Override
    	public int compareTo(FlowBean o) {
    		// 大于返回-1 小于返回 1
    		return this.upFlow>o.upFlow?-1:1;
    	}
    }

*  重新设定`Mapper`的出参`KEYOUT`, 用于排序.

public class FlowCountSort {
    	
    	static class FlowCountSortMapper extends Mapper<LongWritable, Text, FlowBean, Text>{
    
    		FlowBean bean = new FlowBean();
    		Text text = new Text();
    		@Override
    		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, FlowBean, Text>.Context context)throws IOException, InterruptedException {			
    			// <183628945888,12,1>...
    			// 拿到到是上一个统计程序的输出结果, 已经是各手机的总流量信息
    			String line = value.toString();
    			String []fields = line.split(" ");
    			String phoneNumber = fields[0];
    			long upFlow = Long.parseLong(fields[1]);
    			long downFlow = Long.parseLong(fields[2]);
    			
    			// set方法进行赋值
    			bean.setDownFlow(downFlow);
    			bean.setUpFlow(upFlow);
    			
    			//
    			FlowBean newBean = new FlowBean();
    			try {
    				PropertyUtils.copyProperties(newBean, bean);
    			} catch (IllegalAccessException e) {
    				e.printStackTrace();
    			} catch (InvocationTargetException e) {
    				e.printStackTrace();
    			} catch (NoSuchMethodException e) {
    				e.printStackTrace();
    			}
    			
    			text.set(phoneNumber);
    			// <bean,phoneNumber>类型的数据
    			context.write(newBean, text);
    			
    		}
    	}
    	
    	static class FlowCountSortReducer extends Reducer<FlowBean, Text, Text, FlowBean>{
    		@Override
    		//<bean,phoneNumber>
    		protected void reduce(FlowBean bean, Iterable<Text> values,Reducer<FlowBean, Text, Text, FlowBean>.Context context)
    				throws IOException, InterruptedException {
    			context.write(values.iterator().next(), bean);
    		}
    	}
    	
    	public static void main(String []args) throws IOException, ClassNotFoundException, InterruptedException{
    		Configuration conf =  new Configuration();
    //		conf.set("mapreduce.framework.name", "yarn");
    		conf.set("mapreduce.framework.name", "local");
    		conf.set("yarn.resourcemanager.hostname", "localhost");
    		conf.set("fs.defaultFS", "hdfs://localhost:9000/");
    		
    		Job job =  Job.getInstance(conf);
    		
    //		job.setJar("/");
    //		job.setJar("/Users/Sean/Documents/Gitrep/bigdata/hadoop/target/hadoop-0.0.1-SNAPSHOT.jar");
    		// 指定本程序jar包所在地址
    		job.setJarByClass(FlowCountDriver.class);
    		
    		//指定本业务job需要使用的mapper业务类
    		job.setMapperClass(FlowCountSortMapper.class);
    		job.setReducerClass(FlowCountSortReducer.class);
    		
    		// 指定mapper输出数据的kv类型
    		job.setMapOutputKeyClass(FlowBean.class);
    		job.setMapOutputValueClass(Text.class);
    		
    		// 指定最终输出数据的kv类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputKeyClass(FlowBean.class);
    		
    		// 设置分块器
    //		job.setPartitionerClass(ProvincePartitioner.class);
    		// GroupComparator与 ReduceTask的取值 
    		// 1. 现在传入第一个key 2. 随后判断后一个对象key是否相同, 如果相同则将值放入迭代器
    		// 3. 如果不相同,则重新生成一个新的<key, iterator>对传入.
    		job.setPartitionerClass(FlowBeanPartitioner.class);
    		// 设置相同对象的判断器
    		job.setGroupingComparatorClass(FlowBeanComparator.class);
    //		job.setInputFormatClass(cls);
    		// 当数目小于分片数目时候 会报错. 多余时,则有的分片内为空.
    //		job.setNumReduceTasks(5);
    		
    //		// 指定job的输入文件所在目录
    //		FileInputFormat.setInputPaths(job, new Path(args[0]));
    //		// 指定job的输出结果
    //        FileOutputFormat.setOutputPath(job,new Path(args[1]));
    		
    		// 指定job的输入文件所在目录
    		FileInputFormat.setInputPaths(job, new Path("/flowcount/input"));
    		// 指定job的输出结果
    		FileOutputFormat.setOutputPath(job, new Path("/flowcount/output"));
    		
    		Path outPath = new Path("/flowcount/output");
    		FileSystem fs = FileSystem.get(conf);
    		if(fs.exists(outPath)){
    			fs.delete(outPath,true);
    		}
    		// 将job中配置的相关参数, 以及job所用的java类所在的jar包,提交给yarn执行
    //		job.submit();
    		
    		// 
    		boolean flag = job.waitForCompletion(true);
    		System.exit(flag?0:1);
    	}
    }

注: 两者排序方式都可以实现排序的功能, 但是第一种方式会一定程度上消耗`Reducer`的性能, 当数据量较大时, 会导致`Map`的内存容量较大, 不是特别推荐. 一般使用方法2进行排序(我们可以自定义不同的KEY排序的算法).

另外值得一提的是,`Hive`的`Order BY`的排序关键字应当也是使用相同的办法实现的.

--------------------

### Reference ###

\[1\]. [MapReduce使用cleanup()方法实现排序筛选后输出][MapReduce_cleanup]  
\[2\]. [【MR】MapReduce 常见的排序][MR_MapReduce]

[Hadoop MR _ _WordCount]: https://yanxml.blog.csdn.net/article/details/89050203
[https_github.com_SeanYanxml_bigdata]: https://github.com/SeanYanxml/bigdata/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly95YW54bWwuYmxvZy5jc2RuLm5ldA_size_16_color_FFFFFF_t_70]: /images/20220223/35b57455d9c24629995a6a69225ec446.png
[MapReduce_cleanup]: https://blog.csdn.net/ASN_forever/article/details/81354303
[MR_MapReduce]: https://blog.csdn.net/yangjjuan/article/details/78063279