怎么編寫不同MapReudce程序

JDK在序列化的時候，算法會考慮這些事情:

1. 將對象實例相關的類元數據輸出。

2. 遞歸地輸出類的超類描述直到不再有超類。

3. 類元數據完了以后，開始從最頂層的超類開始輸出對象實例的實際數據值.

4. 從上至下遞歸輸出實例的數據

   優(yōu)點:從上面來看java的序列化確實很強大，序列化后得到的信息也很詳細，所以反序列化就變得特別簡單.

所以我們只要implements Serializable接口,JDK會自動處理一切,Java的序列化機制相當復雜，能處理各種對象關系。

        缺點:Java的序列化機制計算量開銷大，且序列化的結果體積太大,有時能達到對象大小的數倍.引用機制也會導致大文件不能分割.

       這些缺點對于Hadoop是非常致命的,因為在Hadoop集群之間需要通訊或者是RPC調用的話,需要序列化，而且要求序列化要快，且體積要小，占用帶寬要小。所以Hadoop就自個玩了一套.

Hadoop的序列化的特點是：

 1 . 緊湊：由于帶寬是集群中信息傳遞的最寶貴的資源所以我們必須想法設法縮小傳遞信息的大小，hadoop的序列化就   為了更好地坐到這一點而設計的。

 2 . 對象可重用：JDK的反序列化會不斷地創(chuàng)建對象，這肯定會造成一定的系統(tǒng)開銷，但是在hadoop的反序列化中，能重復的利用一個對象的readField方法來重新產生不同的對象。

 3 . 可擴展性：Hadoop的序列化有多中選擇

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

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

     PS(網絡來源):hadoop2.X之后是實現一個叫YARN，所有應用（如mapreduce，或者其他spark實時或者離線的計算框架都可以運行在YARN上），YARN還負責對資源的調度等等。YARN的序列化就是用Google開發(fā)的序列化框架protocol Buffers,目前支持支持三種語言C++，java，Python.所以RPC這一層我們就可以利用其他語言來做文章，滿足其他語言開發(fā)者的需求。

接下來的話就是如何使用序列化機制,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);    //模擬分區(qū),存在內存中。
		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);
		//判斷是否在某個分區(qū)中
		Integer index = areaMap.get(prefix);
		if(index!=null){
			area = index;  //如果存在,取相應的數字0,1,2,3
		}
		return area;
	}
}

   2.在Runner中添加兩行代碼.

   3.在Hadoop中的運行結果.