MapReduce编程案例一

2019-08-16

Mapreduce经典案例之GroupingComparator

题目描述

用户在某个位置从某个时刻开始停留了多长时间

现有如下数据文件需要处理
格式：input.csv
字段：用户ID，位置ID，开始时间，停留时长(分钟）
4行样例：
UserA,LocationA,2018-01-01 08:00:00,60
UserA,LocationA,2018-01-01 09:00:00,60
UserA,LocationB,2018-01-01 10:00:00,60
UserA,LocationA,2018-01-01 11:00:00,60
解读：
样例数据中的数据含义是：
用户UserA，在LocationA位置，从8点开始，停留了60分钟
用户UserA，在LocationA位置，从9点开始，停留了60分钟
用户UserA，在LocationB位置，从10点开始，停留了60分钟
用户UserA，在LocationA位置，从11点开始，停留了60分钟

该样例期待输出：
UserA,LocationA,2018-01-01 08:00:00,120
UserA,LocationB,2018-01-01 10:00:00,60
UserA,LocationA,2018-01-01 11:00:00,60

处理逻辑：
1 对同一个用户，在同一个位置，连续的多条记录进行合并
2 合并原则：开始时间取最早时间，停留时长加和

处理逻辑代码

Mapper类

package cn.aparke.mr;

import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
/**
 * @author aparke
 思路分析：

如果参数location参数相同，那么再判断时间（转为long型），对每条数据升序排列，判断当前时间加上停留时间是否与
下一条数据相等，相等则划为一组，通过mapreduce的一个GroupingComparator类来将相同的组的key合并
id作为key无法控制， 所以需要自定义封装类来完成
将key设置为 value1第一个值  
 		  key       			value
map 输入:Object					Text
map 输出:UserLocationBean	NullWritable

逻辑处理：
1、按照userid和locationid分组
2、按照userid和locationid和time排序
 *
 */
public class UserLocationMapper extends Mapper<LongWritable, Text, UserLocationBean, NullWritable> {

	//实例化一个UserLocationBean对象作为输出key值
	UserLocationBean outKey = new UserLocationBean();
	
	@Override
	protected void map(LongWritable key, Text value,Context context)
			throws IOException, InterruptedException {
		
		//数据样例：UserA,LocationA,2018-01-01 08:00:00,60
		String[] split = value.toString().split(",");
		//在UserLocationBean对象中定义set(String [] split)方法，对传进来的数字进行切割
		outKey.set(split);
		//通过key将UserLocation 类封装好的传过去
		context.write(outKey, NullWritable.get());//value传过去为空
	}
}

Reducer类

package cn.aparke.mr;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
/**
 * 
 * @author aparke
 * 		  key       				value
reduce 输入:UserLocationBean		NullWritable		
reduce 输出:UserLocationBean		NullWritable
 */
public class UserLocationReducer extends Reducer<UserLocationBean, NullWritable, UserLocationBean, NullWritable>{

	/**
	 * 传进来的数据key
	 * UserA,LocationA,2018-01-01 08:00:00,60
	 * UserA,LocationA,2018-01-01 09:00:00,60
	 * UserA,LocationB,2018-01-01 10:00:00,60
	 * UserA,LocationA,2018-01-01 11:00:00,60
	 */
	
	//reduce的key的对象
	UserLocationBean outKey = new UserLocationBean();
	//日期解析格式
	SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	@Override
	protected void reduce(UserLocationBean key, Iterable<NullWritable> values,
			Context context)
			throws IOException, InterruptedException {
		
		//遍历每一条传过来的空值value
		//判断如果每一组的key-value的第一条都作为一个key，并设置key
		//判断他们的时间是否连续，如果连续则为同一组数据，然后再设置key
		
		//每次记录与上次记录比较，所以需要实现排序的升序的功能，需要重写UserLocationBean的CompareTo方法
		
		//定义一个计数器
		int count=0;
		for (NullWritable nullWritable : values) {
			count++;
			//如果是这一组key-value中的第一个元素时，直接赋值给outKey对象
			if (count==1) {
				outKey.set(key);
			}else {
				//否则判断停留时间是否连续，如果连续合并停留时间
				long firstTime=0;//第一个时间
				long lastTime=0;//最后一个时间
				//
				try {
					//每次遍历出来得到的时间戳,注意：为当前对象key
					firstTime=simpleDateFormat.parse(key.getTime()).getTime();
					//上一条记录的时间戳和停留时间之和,注意：上次的对象outKey
					//时间戳的单位为毫秒，而停留时长为分钟需单位换算
					lastTime=simpleDateFormat.parse(outKey.getTime()).getTime()+outKey.getDuration()*60*1000; 
							} catch (ParseException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				
				// 如果相等，证明是连续记录，所以合并，更新outKey的停留时长
				if (firstTime == lastTime) {
					// 将两次的停留时长合并，设置为上条数据的停留时长
					outKey.setDuration(outKey.getDuration() + key.getDuration());
				} else {
					// 如果不相等
					// 先输出上一条数据记录
					context.write(outKey, nullWritable);

					// 然后再次记录当前遍历到的这一条记录
					outKey.set(key);
				}
			}
		}
		
		//遍历完后，输出一次
		context.write(outKey, NullWritable.get());//输出的key为对象UserLocationBean ，value为空
	}
}

Job提交类

package cn.aparke.mr;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class UserLocationJobLocalTest {

	public static void main(String[] args) throws Exception {
		//加载配置参数
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		//设置提交的主类
		job.setJarByClass(UserLocationJobLocalTest.class);
		//指定map reduce 类
		job.setMapperClass(UserLocationMapper.class);
		job.setReducerClass(UserLocationReducer.class);
		// 指定maptask的输出类型 可以省去
		job.setMapOutputKeyClass(UserLocationBean.class);
		job.setMapOutputValueClass(NullWritable.class);
		
		// 指定job reducetask的输出类型
		job.setOutputKeyClass(UserLocationBean.class);
		job.setOutputValueClass(NullWritable.class);
		
		//如果连续**（注意，一定连续）**的两条或多条记录满足同组
		//（即compare方法返回0）的条件，那么会把这些key对应的value放在同一个集合中。
		job.setGroupingComparatorClass(UserLocationGC.class);
		
		Path inputPath = new Path("data/input");
		Path outputPath = new Path("data/output");
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outputPath)) {
			fs.delete(outputPath, true);
	}
		// 指定该mapreduce程序数据的输入和输出路径	
		FileInputFormat.setInputPaths(job, inputPath);
		FileOutputFormat.setOutputPath(job, outputPath);
		
		// 提交
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}

UserLocationBean对象类

package cn.aparke.mr;

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

//自己封装的对象类需要实现序列化和反序列化
//如需要将自定义的 bean 放在 key 中传输，则还需要实现 comparable 接口，
//因为 mapreduce 框中的 shuffle 过程一定会对 key 进行排序,此时，自定义的bean 实现的接口应该是：WritableComparable
public class UserLocationBean implements WritableComparable<UserLocationBean>{

	private String userId;
	private String locationId;
	private String time;
	private long duration;//持续时间
	
	//反序列化需要加载无参构造
	public UserLocationBean() {
		// TODO Auto-generated constructor stub
	}
	
	//切分数据设置每个属性
	public void set(String[] split) {

		
		this.setUserId(split[0]);
		this.setLocationId(split[1]);
		this.setTime(split[2]);
		this.setDuration(Long.parseLong(split[3]));
	}
	
	//设置UserLocationBean的set方法
	public void set(UserLocationBean userLocationBean){
		this.setUserId(userLocationBean.getUserId());
		this.setLocationId(userLocationBean.getLocationId());
		this.setTime(userLocationBean.getTime());
		this.setDuration(userLocationBean.getDuration());
	}
	
	@Override
	public String toString() {
//		return userId + "\t" + locationId + "\t" + time + "\t" + duration;
		return userId + "," + locationId + "," + time + "," + duration;
		
	}

	public UserLocationBean(String userId, String locationId, String time, long duration) {
		super();
		this.userId = userId;
		this.locationId = locationId;
		this.time = time;
		this.duration = duration;
	}

	public String getUserId() {
		return userId;
	}

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

	public String getLocationId() {
		return locationId;
	}

	public void setLocationId(String locationId) {
		this.locationId = locationId;
	}

	public String getTime() {
		return time;
	}

	public void setTime(String time) {
		this.time = time;
	}

	public long getDuration() {
		return duration;
	}

	public void setDuration(long duration) {
		this.duration = duration;
	}
	/**
	 * hadoop系统在反序列化该类的对象时要调用的方法
	 */
	public void readFields(DataInput in) throws IOException {
		// TODO Auto-generated method stub
		this.userId = in.readUTF();
		this.locationId = in.readUTF();
		this.time = in.readUTF();
		this.duration = in.readLong();
	}
	/**
	 * hadoop系统在序列化该类的对象时要调用的方法
	 */
	public void write(DataOutput out) throws IOException {
		// TODO Auto-generated method stub
		out.writeUTF(userId);
		out.writeUTF(locationId);
		out.writeUTF(time);
		out.writeLong(duration);
	}
	/**
	 * 排序规则
	 * 
	 * 按照 userId  locationId  和  time 排序  都是 升序
	 *  指定对象排序的方法
	 *  如果指定的数与参数相等返回 0。
	 *  如果指定的数小于参数返回 -1。
	 *  如果指定的数大于参数返回 1。
	 */
	
	public int compareTo(UserLocationBean o) {
		
		// 比较userid是否相等
		int comUserId = this.userId.compareTo(o.userId);
		if (comUserId == 0) {
			// 如果相等，比较locationId
			int comLocationId = this.locationId.compareTo(o.locationId);
			if (comLocationId == 0) {
				// 如果locationid也相等，再比较time
				int comTime = this.time.compareTo(o.time);
				if (comTime == 0) {
					// 如果time相等，则返回0
					return 0;
				} else {
					return comTime > 0 ? 1 : -1;
				}
			} else {
				return comLocationId > 0 ? 1 : -1;
			}
		} else {
			return comUserId > 0 ? 1 : -1;
		}
	}
}

GroupingComparator类

描述
GroupingComparator是在reduce阶段分组来使用的，由于reduce阶段，如果key相同的一组，只取第一个key作为key，迭代所有的values。
如果reduce的key是自定义的bean，我们只需要bean里面的某个属性相同就认为这样的key是相同的，这是我们就需要之定义GroupCoparator来“欺骗”reduce了。我们需要理清楚的还有map阶段你的几个自定义：
parttioner中的getPartition（）这个是map阶段自定义分区， bean中定义CopmareTo()是在溢出和merge时用来来排序的。

我们就需要使用GroupingComparatorClass来自定义分组方式。我们需要定义一个Comparator函数，令其继承WritableComparator，并重写compare方法。
在compare方法方法中，我们定义规约器的key分组方式。
参考文章：
https://www.cnblogs.com/jingpeng77/p/10098847.html
https://www.jianshu.com/p/b36d8c890e5c

package cn.aparke.mr;

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

public class UserLocationGC extends WritableComparator{

	public UserLocationGC() {
		super(UserLocationBean.class, true);
	}

	@Override
	public int compare(WritableComparable a, WritableComparable b) {
		// 强转 将WritableComparable转为UserLocationBean
		UserLocationBean bean_a = (UserLocationBean) a;
		UserLocationBean bean_b = (UserLocationBean) b;

		return bean_a.getUserId().compareTo(bean_b.getUserId());
	}
}

项目地址

Mapreduce经典案例一