Hadoop MapReduce编程 API入门系列之网页流量版本1（二十一）

　　不多说，直接上代码。

　　对流量原始日志进行流量统计，将不同省份的用户统计结果输出到不同文件。

代码

package zhouls.bigdata.myMapReduce.areapartition;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

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

public class FlowBean implements WritableComparable<FlowBean>{


private String phoneNB;
private long up_flow;
private long d_flow;
private long s_flow;

//在反序列化时，反射机制需要调用空参构造函数，所以显示定义了一个空参构造函数
public FlowBean(){}

//为了对象数据的初始化方便，加入一个带参的构造函数
public FlowBean(String phoneNB, long up_flow, long d_flow) {
this.phoneNB = phoneNB;
this.up_flow = up_flow;
this.d_flow = d_flow;
this.s_flow = up_flow + d_flow;
}

public String getPhoneNB() {
return phoneNB;
}

public void setPhoneNB(String phoneNB) {
this.phoneNB = phoneNB;
}

public long getUp_flow() {
return up_flow;
}

public void setUp_flow(long up_flow) {
this.up_flow = up_flow;
}

public long getD_flow() {
return d_flow;
}

public void setD_flow(long d_flow) {
this.d_flow = d_flow;
}

public long getS_flow() {
return s_flow;
}

public void setS_flow(long s_flow) {
this.s_flow = s_flow;
}

//将对象数据序列化到流中
public void write(DataOutput out) throws IOException {

out.writeUTF(phoneNB);
out.writeLong(up_flow);
out.writeLong(d_flow);
out.writeLong(s_flow);

}

//从数据流中反序列出对象的数据
//从数据流中读出对象字段时，，必须跟序列化时的顺序保持一致
public void readFields(DataInput in) throws IOException {

phoneNB = in.readUTF();
up_flow = in.readLong();
d_flow = in.readLong();
s_flow = in.readLong();

}


@Override
public String toString() {

return "" + up_flow + "\t" +d_flow + "\t" + s_flow;
}

public int compareTo(FlowBean o) {
return s_flow>o.getS_flow()?-1:1;
}

}

package zhouls.bigdata.myMapReduce.areapartition;

import java.util.HashMap;

import org.apache.hadoop.mapreduce.Partitioner;

public class AreaPartitioner<KEY, VALUE> extends Partitioner<KEY, VALUE>{

private static HashMap<String,Integer> areaMap = new HashMap<>();

static{
areaMap.put("135", 0);
areaMap.put("136", 1);
areaMap.put("137", 2);
areaMap.put("138", 3);
areaMap.put("139", 4);
}





@Override
public int getPartition(KEY key, VALUE value, int numPartitions) {
//从key中拿到手机号，查询手机归属地字典，不同的省份返回不同的组号

int areaCoder = areaMap.get(key.toString().substring(0, 3))==null?5:areaMap.get(key.toString().substring(0, 3));

return areaCoder;
}

}

package zhouls.bigdata.myMapReduce.areapartition;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import zhouls.bigdata.myMapReduce.areapartition.FlowBean;

/**
* 对流量原始日志进行流量统计，将不同省份的用户统计结果输出到不同文件
* 需要自定义改造两个机制：
* 1、改造分区的逻辑，自定义一个partitioner
* 2、自定义reduer task的并发任务数
*
*
*
*/
public class FlowSumArea implements Tool {