知之為知之，不知為不知

Hadoop序列化

序列化概念

? 序列化就是把記憶體中的物件，轉換成位元組序列（或其他資料傳輸協議）以便于存盤（持久化）和網路傳輸反序列化就是將收到位元組序列（或其他資料傳輸協議）或者是硬碟的持久化資料，轉換成記憶體中的物件，

序列化作用

? “活的”物件只生存在記憶體里，關機斷電就沒有了，而且“活的”物件只能由本地的行程使用，不能被發送到網路上的另外一臺計算機，然而序列化可以存盤“活的”物件，可以將“活的”物件發送到遠程計算機

為什么不用Java的序列化

? Java的序列化是一個重量級序列化框架 Serializable，一個物件被序列化后，會附帶很多額外的資訊（各種校驗資訊， Header，繼承體系等），不便于在網路中高效傳輸，所以， Hadoop自己開發了一套序列化機制（ Writable），
Hadoop序列化特點

（1）緊湊：高效使用存盤空間，
（2）快速：讀寫資料的額外開銷小，
（3）可擴展：隨著通信協議的升級而可升級
（4）互操作：支持多語言的互動

常用資料的序列化型別

自定義bean物件實作序列化介面步驟

（1）必須實作Writable介面

（2）空參建構式

public FlowBean() {	super();}

（3）重寫序列化方法

@Overridepublic void write(DataOutput out) throws IOException {	out.writeLong(upFlow);	out.writeLong(downFlow);	out.writeLong(sumFlow);}

（4）重寫反序列化方法

@Overridepublic void readFields(DataInput in) throws IOException {	upFlow = in.readLong();	downFlow = in.readLong();	sumFlow = in.readLong();}

（5）方法順序一致

注意反序列化的順序和序列化的順序完全一致

（6）重寫toString

要想把結果顯示在檔案中，需要重寫toString()，可用”\t”分開，方便后續用，

（7）實作Comparable介面

如果需要將自定義的bean放在key中傳輸，則還需要實作Comparable介面，因為MapReduce框中的Shuffle程序要求對key必須能排序，

@Overridepublic int compareTo(FlowBean o) {	// 倒序排列，從大到小	return this.sumFlow > o.getSumFlow() ? -1 : 1;}

序列化案例實操

需求

統計每一個手機號耗費的總上行流量、下行流量、總流量

輸出資料

13560436666 		1116		      954 			2070
手機號碼		    上行流量        下行流量		總流量

需求分析

Bean代碼

// 1 實作writable介面
public class FlowBean implements Writable{

	private long upFlow ;
	private long downFlow;
	private long sumFlow;
	
	//2  反序列化時，需要反射呼叫空參建構式，所以必須有
	public FlowBean() {
		super();
	}

	public FlowBean(long upFlow, long downFlow) {
		super();
		this.upFlow = upFlow;
		this.downFlow = downFlow;
		this.sumFlow = upFlow + downFlow;
	}
	
	//3  寫序列化方法
	@Override
	public void write(DataOutput out) throws IOException {
		out.writeLong(upFlow);
		out.writeLong(downFlow);
		out.writeLong(sumFlow);
	}
	
	//4 反序列化方法
	//5 反序列化方法讀順序必須和寫序列化方法的寫順序必須一致
	@Override
	public void readFields(DataInput in) throws IOException {
		this.upFlow  = in.readLong();
		this.downFlow = in.readLong();
		this.sumFlow = in.readLong();
	}

	// 6 撰寫toString方法，方便后續列印到文本
	@Override
	public String toString() {
		return upFlow + "\t" + downFlow + "\t" + sumFlow;
	}

	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;
	}
}

Mapper代碼

public class FlowCountMapper extends Mapper<LongWritable, Text, Text, FlowBean>{
	
	FlowBean v = new FlowBean();
	Text k = new Text();
	
	@Override
	protected void map(LongWritable key, Text value, Context context)	throws IOException, InterruptedException {
		
		// 1 獲取一行
		String line = value.toString();
		
		// 2 切割欄位
		String[] fields = line.split("\t");
		
		// 3 封裝物件
		// 取出手機號碼
		String phoneNum = fields[1];
		// 取出上行流量和下行流量
		long upFlow = Long.parseLong(fields[fields.length - 3]);
		long downFlow = Long.parseLong(fields[fields.length - 2]);

		k.set(phoneNum);
		v.set(downFlow, upFlow);
		
		// 4 寫出
		context.write(k, v);
	}
}

Reducer代碼

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

	@Override
	protected void reduce(Text key, Iterable<FlowBean> values, Context context)throws IOException, InterruptedException {

		long sum_upFlow = 0;
		long sum_downFlow = 0;

		// 1 遍歷所用bean，將其中的上行流量，下行流量分別累加
		for (FlowBean flowBean : values) {
			sum_upFlow += flowBean.getUpFlow();
			sum_downFlow += flowBean.getDownFlow();
		}

		// 2 封裝物件
		FlowBean resultBean = new FlowBean(sum_upFlow, sum_downFlow);
		
		// 3 寫出
		context.write(key, resultBean);
	}
}

Driver代碼

public class FlowsumDriver {

	public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
		
// 輸入輸出路徑需要根據自己電腦上實際的輸入輸出路徑設定
args = new String[] { "e:/input/inputflow", "e:/output1" };

		// 1 獲取配置資訊，或者job物件實體
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);

		// 6 指定本程式的jar包所在的本地路徑
		job.setJarByClass(FlowsumDriver.class);

		// 2 指定本業務job要使用的mapper/Reducer業務類
		job.setMapperClass(FlowCountMapper.class);
		job.setReducerClass(FlowCountReducer.class);

		// 3 指定mapper輸出資料的kv型別
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(FlowBean.class);

		// 4 指定最終輸出的資料的kv型別
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);
		
		// 5 指定job的輸入原始檔案所在目錄
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		// 7 將job中配置的相關引數，以及job所用的java類所在的jar包， 提交給yarn去運行
		boolean result = job.waitForCompletion(true);
		System.exit(result ? 0 : 1);
	}
}

相關資料

本文配套GitHub：https://github.com/zhutiansama/FocusBigData

標籤：Java

上一篇：單例模式和工廠模式

下一篇：想做時間管理大師？你可以試試Mybatis Plus代碼生成器