怎么编写不同MapReudce程序

JDK在序列化的时候，算法会考虑这些事情:

1. 将对象实例相关的类元数据输出。

2. 递归地输出类的超类描述直到不再有超类。

3. 类元数据完了以后，开始从最顶层的超类开始输出对象实例的实际数据值.

4. 从上至下递归输出实例的数据

   优点:从上面来看java的序列化确实很强大，序列化后得到的信息也很详细，所以反序列化就变得特别简单.

所以我们只要implements Serializable接口,JDK会自动处理一切,Java的序列化机制相当复杂，能处理各种对象关系。

        缺点:Java的序列化机制计算量开销大，且序列化的结果体积太大,有时能达到对象大小的数倍.引用机制也会导致大文件不能分割.

       这些缺点对于Hadoop是非常致命的,因为在Hadoop集群之间需要通讯或者是RPC调用的话,需要序列化，而且要求序列化要快，且体积要小，占用带宽要小。所以Hadoop就自个玩了一套.

Hadoop的序列化的特点是：

 1 . 紧凑：由于带宽是集群中信息传递的最宝贵的资源所以我们必须想法设法缩小传递信息的大小，hadoop的序列化就   为了更好地坐到这一点而设计的。

 2 . 对象可重用：JDK的反序列化会不断地创建对象，这肯定会造成一定的系统开销，但是在hadoop的反序列化中，能重复的利用一个对象的readField方法来重新产生不同的对象。

 3 . 可扩展性：Hadoop的序列化有多中选择

             a.可以利用实现hadoop框架中的Writable接口。(原生的)

             b.使用开源的序列化框架protocol Buffers，Avro等框架。

     PS(网络来源):hadoop2.X之后是实现一个叫YARN，所有应用（如mapreduce，或者其他spark实时或者离线的计算框架都可以运行在YARN上），YARN还负责对资源的调度等等。YARN的序列化就是用Google开发的序列化框架protocol Buffers,目前支持支持三种语言C++，java，Python.所以RPC这一层我们就可以利用其他语言来做文章，满足其他语言开发者的需求。

接下来的话就是如何使用序列化机制,Writable介绍如下.

1. 定义可序列化的JavaBean.com.codewatching.fluxcount.bean.FlowBean

package com.codewatching.fluxcount.bean;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.Writable;
public class FlowBean implements Writable {
	private String phoneNum;
	private long upFlow;
	private long downFlow;
	private long sumFlow;
	public FlowBean(){}
	public FlowBean(String phoneNum, long upFlow, long downFlow) {
		super();
		this.phoneNum = phoneNum;
		this.upFlow = upFlow;
		this.downFlow = downFlow;
		this.sumFlow = upFlow+downFlow;
	}
	public String getPhoneNum() {
		return phoneNum;
	}
	public void setPhoneNum(String phoneNum) {
		this.phoneNum = phoneNum;
	}
	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 long getSumFlow() {
		return sumFlow;
	}

	public void setSumFlow(long sumFlow) {
		this.sumFlow = sumFlow;
	}
	@Override
	public void write(DataOutput out) throws IOException {
		out.writeUTF(phoneNum);
		out.writeLong(downFlow);
		out.writeLong(upFlow);
		out.writeLong(sumFlow);
	}
	@Override
	public void readFields(DataInput in) throws IOException {
		phoneNum = in.readUTF();
		downFlow = in.readLong();
		upFlow = in.readLong();
		sumFlow = in.readLong();
	}
	@Override
	public String toString() {
		return upFlow+"\t"+downFlow+"\t"+sumFlow;
	}
}

       2. 编写Mapper,Reducer,Runner.

package com.codewatching.fluxcount.hadoop;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import com.codewatching.fluxcount.bean.FlowBean;
public class FlowSumMapper extends Mapper<LongWritable, Text, Text, FlowBean>{
	@Override
	protected void map(LongWritable key, Text value,Context context)
			throws IOException, InterruptedException {
		String line = value.toString();
		String[] fileds = line.split("\t");
		int length = fileds.length;
		String phoneNum = fileds[1];
		long upFlow = Long.parseLong(fileds[length-3]);
		long downFlow = Long.parseLong(fileds[length-2]);
		FlowBean flowBean = new FlowBean(phoneNum, upFlow, downFlow);
		//以flowBean为value供reducer处理
		context.write(new Text(phoneNum), flowBean);
	}
}

package com.codewatching.fluxcount.hadoop;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import com.codewatching.fluxcount.bean.FlowBean;
public class FlowSumReducer extends Reducer<Text, FlowBean, Text, FlowBean>{
	@Override
	protected void reduce(Text key, Iterable<FlowBean> values,Context context)
			throws IOException, InterruptedException {
		long _downFlow = 0;
		long _upFlow = 0;
		for (FlowBean flowBean : values) {
			_downFlow += flowBean.getDownFlow();
			_upFlow += flowBean.getUpFlow();
		}
		FlowBean bean = new FlowBean(key.toString(), _upFlow, _downFlow);
		context.write(key, bean);
	}
}

package com.codewatching.fluxcount.hadoop;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import com.codewatching.fluxcount.bean.FlowBean;
public class FlowSumRunner extends Configured implements Tool{
	@Override
	public int run(String[] args) throws Exception {
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);
		configuration.set("mapreduce.job.jar", "fluxcount.jar");
		job.setJarByClass(FlowSumRunner.class);
		job.setMapperClass(FlowSumMapper.class);
		job.setReducerClass(FlowSumReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileSystem fileSystem = FileSystem.get(configuration);
		Path path = new Path(args[1]);
		if(fileSystem.exists(path)){
			fileSystem.delete(path, true);
		}
		FileOutputFormat.setOutputPath(job, path);
		
		return job.waitForCompletion(true)?0:1;
	}
	public static void main(String[] args) throws Exception {
		ToolRunner.run(new Configuration(), new FlowSumRunner(), args);
	}
}

1. 在案例的基础上，编写一个Partitionner实现类.

package com.codewatching.fluxcount.hadoop;
import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
import com.codewatching.fluxcount.bean.FlowBean;
public class AreaPartitioner extends Partitioner<Text, FlowBean>{
	private static Map<String,Integer> areaMap;
	static{
		areaMap = new HashMap<String, Integer>();
		areaMap.put("135", 0);    //模拟分区,存在内存中。
		areaMap.put("137", 1);
		areaMap.put("138", 2);
		areaMap.put("139", 3);
	}
	@Override
	public int getPartition(Text key, FlowBean value, int numPartitions) {
		int area = 4;  //默认都是为4
		String prefix = key.toString().substring(0,3);
		//判断是否在某个分区中
		Integer index = areaMap.get(prefix);
		if(index!=null){
			area = index;  //如果存在,取相应的数字0,1,2,3
		}
		return area;
	}
}

   2.在Runner中添加两行代码.

   3.在Hadoop中的运行结果.