MapReduce编程案例六

2019-08-23

Mapreduce经典案例之分组TopN

题目描述

统计每一个订单中成交金额最大的三笔

数据样例：
order001,u001,小米6,1999.9,2
order001,u001,雀巢咖啡,99.0,2
order001,u001,安慕希,250.0,2
order001,u001,经典红双喜,200.0,4
order001,u001,防水电脑包,400.0,2
order002,u002,小米手环,199.0,3
order002,u002,榴莲,15.0,10
order002,u002,苹果,4.5,20
order002,u002,肥皂,10.0,40

处理逻辑：
1 map：读取数据切分字段，使用orderId作为key，封装数据到一个bean中作为map端的value传输，在对象bean中实现WritableComparable接口，覆盖compareTo方法，重写比较规则，比较金额倒序，如果金额相同则比较商品名字
2 reduce： reduce task提供的values迭代器，每次迭代返回给我们的都是同一个对象，在迭代器（循坏遍历中）每次遍历构造一个新的对象，来存储本次迭代出来的值，将很多个对象存入到list中
3 通过context传进来的参数控制topn

处理逻辑代码

Mapper类

package cn.aparke.mr.order.topn;

import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class OrderTopnMapper extends Mapper<LongWritable, Text, Text, OrderBean> {

	OrderBean orderBean = new OrderBean();
	Text outKey = new Text();
	
	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, OrderBean>.Context context)
			throws IOException, InterruptedException {
		String[] filds = value.toString().split(",");//按逗号切分数据
		
		//通过OderBean中的set方法一次性设置每个对象参数
		orderBean.set(filds[0], filds[1], filds[2], Float.parseFloat(filds[3]), Integer.parseInt(filds[4]));
		/*
		 * 如果在这里new一个oderBean 和text 每一个MapTask每读一行就要new一次，会产生大量对象
		所以将新建对象提到外面去，使用同一个对象的引用，值却不是相同的值
		//OrderBean orderBean = new OrderBean();
		//Text k = new Text();
		*/
		outKey.set(filds[0]);//userId作为key
		context.write(outKey, orderBean);
	}

}

Reducer类

在日常开发中，很多时候都需要对一些数据进行排序的操作。然而那些数据一般都是放在一个集合中如：Map ，Set ，List 等集合中。他们都提共了一个排序方法 sort()，要对数据排序直接使用这个方法就行，但是要保证集合中的对象是可比较的。
怎么让一个对象是可比较的，那就需要该对象实现 Comparable 接口啦。然后重写里面的compareTo()方法。我们可以看到Java中很多类都是实现类这个接口的如：Integer，Long 等等

package cn.aparke.mr.order.topn;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class OrderTopnReducer extends Reducer<Text, OrderBean, OrderBean, NullWritable> {

	@Override
	protected void reduce(Text key, Iterable<OrderBean> values,
			Reducer<Text, OrderBean, OrderBean, NullWritable>.Context context) throws IOException, InterruptedException {

		int topn = context.getConfiguration().getInt("top.n", 2);
		ArrayList<OrderBean> beanList = new ArrayList<>();
		//迭代器
		// reduce task提供的values迭代器，每次迭代返回给我们的都是同一个对象，只是set了不同的值
		for (OrderBean orderBean : values) {
			
			// 每次遍历构造一个新的对象，来存储本次迭代出来的值
			OrderBean newBean = new OrderBean();
			newBean.set(orderBean.getOrderId(), orderBean.getUserId(), orderBean.getPdtName(), orderBean.getPrice(), orderBean.getNumber());
			// 将很多个对象存入到list中
			beanList.add(newBean);
		}
		
		// 对beanList中的orderBean对象排序实现WritableComparable接口（按总金额大小倒序排序,如果总金额相同，则比商品名称）
		Collections.sort(beanList);//sort排序的规则，须在他的对象类中实现WritableComparable接口 重写覆盖compareto方法
		//打印topn
		for (int i=0;i<topn;i++) {
			context.write(beanList.get(i), NullWritable.get());//打印前n个beanlist对象
		}
	}
}

JobDrivcer驱动提交类

package cn.aparke.mr.order.topn;

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.NullWritable;
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;


public class OrderTopnDriver {
	public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();
		conf.setInt("top.n" , 3);// 通过代码设置参数
		
		Job job = Job.getInstance(conf);

		job.setJarByClass(OrderTopnDriver.class);

		job.setMapperClass(OrderTopnMapper.class);
		job.setReducerClass(OrderTopnReducer.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(OrderBean.class);
		
		job.setOutputKeyClass(OrderBean.class);
		job.setOutputValueClass(NullWritable.class);
		
		job.setNumReduceTasks(2);
		//此处如果设置ReduceTask数为2，则一定会把1，3分为一组，2为一组
		//因为系统缺省的Partitioner是HashPartitioner，它以key的Hash值对Reducer的数目取模，
		//得到对应的Reducer。这样保证如果有相同的key值，肯定被分配到同一个reducre上。
		//由三个key 3%2=1 1%2=1 故1，3被分在一起
		Path inpath=new Path("d:/testdata/order/input");
		Path outpath=new Path("d:/testdata/order/output");
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outpath)) {
			fs.delete(outpath, true);
		}
		
		FileInputFormat.setInputPaths(job, inpath);
		FileOutputFormat.setOutputPath(job, outpath);
		
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion?0:1);
	}
}

OrderBean对象类

package cn.aparke.mr.order.topn;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;

public class OrderBean implements WritableComparable<OrderBean> {

	private String orderId;
	private String userId;
	private String pdtName;
	private float price;
	private int number;
	private float amountFee;// 销售总金额

	// 此方法用来设置对象的参数
	public void set(String orderId, String userId, String pdtName, float price, int number) {
		this.orderId = orderId;
		this.userId = userId;
		this.pdtName = pdtName;
		this.price = price;
		this.number = number;
		this.amountFee = number * price;// 销售金额等于数量*单价
	}

	public String getOrderId() {
		return orderId;
	}

	public void setOrderId(String orderId) {
		this.orderId = orderId;
	}

	public String getUserId() {
		return userId;
	}

	public void setUserId(String userId) {
		this.userId = userId;
	}

	public String getPdtName() {
		return pdtName;
	}

	public void setPdtName(String pdtName) {
		this.pdtName = pdtName;
	}

	public float getPrice() {
		return price;
	}

	public void setPrice(float price) {
		this.price = price;
	}

	public int getNumber() {
		return number;
	}

	public void setNumber(int number) {
		this.number = number;
	}

	public float getAmountFee() {
		return amountFee;
	}

	public void setAmountFee(float amountFee) {
		this.amountFee = amountFee;
	}

	@Override
	public String toString() {
		return this.orderId + "," + this.userId + "," + this.pdtName + "," + this.price + "," + this.number + ","
				+ this.amountFee;
	}

	// 序列化
	@Override
	public void write(DataOutput out) throws IOException {
		out.writeUTF(this.orderId);
		out.writeUTF(this.userId);
		out.writeUTF(this.pdtName);
		out.writeFloat(this.price);
		out.writeInt(this.number);
	}

	// 反序列化
	@Override
	public void readFields(DataInput in) throws IOException {
		this.orderId = in.readUTF();
		this.userId = in.readUTF();
		this.pdtName = in.readUTF();
		this.price = in.readFloat();
		this.number = in.readInt();
		this.amountFee = this.price * this.number;// 序列化中虽然没有序列化出，但是反序列化可以通过逻辑实现进去
	}

	// 比较规则，既然是求topN 那么就应该是倒序排列。先比总金额，如果相同，再比商品名称
	@Override
	public int compareTo(OrderBean o) {
		//三元表达式实现比较规则
		return Float.compare(o.getAmountFee(), this.amountFee) == 0 ? o.getPdtName().compareTo(this.pdtName)
				: Float.compare(o.getAmountFee(), this.amountFee);
	}
}

项目地址

Mapreduce经典案例