MapReduce编程案例二

2019-08-18

Mapreduce经典案例之Partitioner

题目描述

统计不同归属地的用户上网流量

解读：
一组数据中是样例数据中属性分别是
时间戳手机号 MAC地址 ip地址访问网站域名上行数据包下行数据包上行流量下行流量访问响应状态码

处理逻辑：
1 先切分出需要的数据列第二列以及倒数第二三列。
2 通过Partitioner类，实现getPartition方法，在Mapper阶段，当Mapper处理好数据后，这些数匀的分据需要经过Partitioner进行分区，来选择不同的Reducer处理，从而将Mapper的输出结果按自定义分区在Reducer上面执行，然后设置Reduce个数job.setNumReduceTasks();将不同类型结果输出到不同文件中。

处理逻辑代码

Mapper类

package cn.aparke.mr.flow;

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

/**
 * 需求说明 现有如下数据 1363157985066 13726230503 00-FD-07-A4-72-B8:CMCC 120.196.100.82 i02.c.aliimg.com 24 27 2481 24681 200 
 * 分别为 ：时间戳 	手机号 	mac地址  ip	网址 上下行数据包    上下行流量 及响应状态
 * 通过以上数据输出一个人的手机号为key value 输出上下总流量 下行总流量 以及总流量
 * 
 * key 手机号 value 上行流量 下行流量 总流量
 * 
 * map	 输入 LongWritable Text 
 * 		 输出 Text FlowBean
 */
public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean> {

	Text keyout = new Text();
	FlowBean flowBean = new FlowBean();

	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context)
			throws IOException, InterruptedException {
		String[] fields = value.toString().split("\t");
		String phone = fields[1];//手机号
		
		String upFlow = fields[fields.length-3];//拿到上行流量
		String downFlow = fields[fields.length-2];//拿到下行流量
		
		flowBean.setUpFlow(Integer.parseInt(upFlow));
		flowBean.setDownFlow(Integer.parseInt(downFlow));
		
		keyout.set(phone);
		context.write(keyout, flowBean);
		
	}
}

Reducer类

package cn.aparke.mr.flow;

import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean>{

	FlowBean outValue=new FlowBean();
	@Override
	protected void reduce(Text key, Iterable<FlowBean> values,Context context)
			throws IOException, InterruptedException {
		int upSum=0;
		int downSum=0;
		
		for (FlowBean value : values) {
			upSum+=value.getUpFlow();
			downSum+=value.getDownFlow();
		}
		outValue.setUpFlow(upSum);
		outValue.setDownFlow(downSum);
		outValue.setAmount(upSum, downSum);
		
		context.write(key, outValue);
	}
}

Job提交类

package cn.aparke.mr.flow;

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.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 JobSumbitToLocal {
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf,"流量统计");
		
		job.setJarByClass(JobSumbitToLocal.class);
		job.setMapperClass(FlowMapper.class);
		job.setReducerClass(FlowReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);
		
		Path inpath=new Path("d:/testdata/flow/input");
		Path outpath=new Path("d:/testdata/flow/outputplus");
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outpath)) {
			fs.delete(outpath, true);
		}
		
		//实现自定义分区必须有下面两行代码，因为设置了有六个分区器需要设置六个reduceTask
		//设置参数：maptask在做数据分区时，用哪个分区逻辑类  （如果不指定，它会用默认的HashPartitioner）
//		设置自定义分区规则
		job.setPartitionerClass(FlowPartitioner.class);
//		设置Reduce个数
		job.setNumReduceTasks(6);
		
		FileInputFormat.setInputPaths(job, inpath);
		FileOutputFormat.setOutputPath(job, outpath);
		
		boolean waitForCompletion = job.waitForCompletion(true);
		
		System.exit(waitForCompletion?0:1);
	}
}

FlowBean对象类

package cn.aparke.mr.flow;

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

/*
 * /**
 * 自定义复杂数据类型：
 *  (1)需要实现Writable接口
 *  (2)需要实现接口中的write和readFields方法
 *  (3)比较隐蔽的一点,需要定义默认的空构造方法
 *  如果是key使用自定义类型，就必须覆盖CompareTo方法！value则不用。
 *  属性必须覆盖getter/setter方法!
 */

public class FlowBean implements Writable {

	private int upFlow;//上行流量
	private int downFlow;//下行流量
	private int amount;//总流量
	
//	public FlowBean() {
//		// TODO Auto-generated constructor stub
//	}

	public int getAmount() {
		return amount;
	}

	public void setAmount(int upFlow, int downFlow) {
		this.amount = upFlow+downFlow;
	}

	@Override
	public String toString() {
//		return "FlowBean [upFlow=" + upFlow + ", downFlow=" + downFlow + ", amount=" + amount + "]";
		return upFlow + "\t" + downFlow + "\t" +amount;
	}

	public int getUpFlow() {
		return upFlow;
	}
	public void setUpFlow(int upFlow) {
		this.upFlow = upFlow;
	}
	public int getDownFlow() {
		return downFlow;
	}
	public void setDownFlow(int downFlow) {
		this.downFlow = downFlow;
	}
	
	@Override
	public void readFields(DataInput in) throws IOException {
		this.upFlow = in.readInt();
		this.downFlow = in.readInt();
		this.amount = in.readInt();
	}

	@Override
	public void write(DataOutput out) throws IOException {
		out.writeInt(upFlow);
		out.writeInt(downFlow);
		out.writeInt(amount);
	}
	
}

Partitioner类

描述
Partitioner决定了Map Task 输出的每条数据交给哪个Reduce Task 来处理。Partitioner 有两个功能：
１) 均衡负载。它尽量将工作均匀地分配给不同的 Reduce。
２) 效率。它的分配速度一定要非常快。

Partitioner 的默认实现：hash(key) mod R，这里的R代表Reduce Task 的数目，意思就是对key进行hash处理然后取模。很多情况下，用户需要自定义 Partitioner，比如“hash(hostname(URL)) mod R”，它确保相同域名下的网页交给同一个 Reduce Task 来处理。用户自定义Partitioner，需要继承Partitioner类，实现它提供的一个方法。

自定义partitioner
每一个Reduce的输出都是有序的，但是将所有Reduce的输出合并到一起却并非是全局有序的，如果要做到全局有序，我们该怎么做呢？最简单的方式，只设置一个Reduce task，但是这样完全发挥不出集群的优势，而且能应对的数据量也很受限。最佳的方式是自己定义一个Partitioner，用输入数据的最大值除以系统Reduce task数量的商作为分割边界，也就是说分割数据的边界为此商的1倍、2倍至numPartitions-1倍，这样就能保证执行partition后的数据是整体有序的。

解决数据倾斜
另一种需要我们自己定义一个Partitioner的情况是各个Reduce task处理的键值对数量极不平衡。对于某些数据集，由于很多不同的key的hash值都一样，导致这些键值对都被分给同一个Reducer处理，而其他的Reducer处理的键值对很少，从而拖延整个任务的进度。当然，编写自己的Partitioner必须要保证具有相同key值的键值对分发到同一个Reducer。

自定义的Key包含了好几个字段，比如自定义key是一个对象，包括type1，type2，type3,只需要根据type1去分发数据，其他字段用作二次排序。

package cn.aparke.mr.flow;

import java.util.HashMap;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

//需求：将统计结果按照手机号的前缀进行区分,写到不同的文件中。
/*
自定义FlowPartitioner类实现 ，继承Partitioner类 重写方法getPartition
定义MapReduce自定义分区规则
Driver参数
设置自定义分区规则
job.setPartitionerClass(FlowPartitioner.class);
设置Reduce个数
job.setNumReduceTasks(3);
*/

/**
 * 输入是Map的结果对<key, value>和Reducer的数目，输出则是分配的Reducer（整数编号）。
 * 就是指定Mappr输出的键值对到哪一个reducer上去。
 * 系统缺省的Partitioner是HashPartitioner，它以key的Hash值对Reducer的数目取模，
 * 得到对应的Reducer。这样保证如果有相同的key值，肯定被分配到同一个reducre上。
 * 如果有N个reducer，编号就为0,1,2,3……(N-1)。

 * @author aparke
 *
 */
public class FlowPartitioner extends Partitioner<Text, FlowBean>{

	//定义一个hashMap 通过手机号前缀的不同放置到不同的分区编号
	//此处只是简单模拟使用partition的方法，实际生产中我们可以根据不同地区的手机前几位不同我们可以将数据归到不同的归属地
	//此处是不是用静态都行
	static HashMap<String, Integer> mapCount=new HashMap<>();
	static {
		mapCount.put("135", 0);
		mapCount.put("136", 1);
		mapCount.put("137", 2);
		mapCount.put("138", 3);
		mapCount.put("139", 4);
	}
	
	//传进来的是map的key value 通过自己设置分区器数量
	@Override
	public int getPartition(Text key, FlowBean value, int numPartitions) {
		
		Integer parts = mapCount.get(key.toString().substring(0, 3));//拿到key的前三位，通过hashmap得到对应的所属分区
		if (parts!=null) {//如果没有匹配的则放到编号为5的分区
			return parts;
		}else {
			return 5;
		}
	//方法二：通过string的startsWith方法匹配字符串开始元素，匹配到设置分区
	 /* 
	 	if (key.toString().startsWith("135")){
            return 0;
        }else if(key.toString().startsWith("136")){
            return 1;
        }else {
            return 2;
        }
		*/
	}
}

项目地址

Mapreduce经典案例一