2019獨角獸企業重金招聘Python工程師標準>>>

hadoop datajoin 之reduce side join

hadoop提供了一個叫datajoin的jar包，用于解決兩張表關聯的問題。jar位于/hadoop/contrib/datajoin將jar包引入進行開發。

涉及的幾個概念:

1.Data Source：基本與關系數據庫中的表相似，形式為：（例子中為CSV格式）

　　　　　　Customers　　　　　　　　　　　　　　　　 ?Orders
　　　　　　1,Stephanie Leung,555-555-5555　　　　　　3,A,12.95,02-Jun-2008
　　　　　　2,Edward Kim,123-456-7890　　　　　　　　?1,B,88.25,20-May-2008
　　　　　　3,Jose Madriz,281-330-8004　　　　　　　　?2,C,32.00,30-Nov-2007
　　　　　　4,David Stork,408-555-0000　　　　　　　　 ?3,D,25.02,22-Jan-2009

2.Tag：由于記錄類型（Customers或Orders）與記錄本身分離，標記一個Record會確保特殊元數據會一致存在于記錄中。在這個目的下，我們將使用每個record自身的Data source名稱標記每個record。

3.Group Key：Group Key類似于關系數據庫中的鏈接鍵（join key），在我們的例子中，group key就是Customer ID（第一列的3）。由于datajoin包允許用戶自定義group key，所以其較之關系數據庫中的join key更一般、平常。

使用以下樣例數據

customers-20140716

1,Stephanie Leung,555-555-5555
2,Edward Kim,123-456-7890
3,Jose Madriz,281-330-8004
4,David Stork,408-555-0000

orders-20140716

3,A,12.95,02-Jun-2008
1,B,88.25,20-May-2008
2,C,32.00,30-Nov-2007
3,D,25.02,22-Jan-2009
請看流程圖

第一部分，自定義的數據類型。數據類型主要包含兩部分tag和data，tag是給數據打上的標簽用來表示數據來源于哪個文件。data是數據記錄。

上代碼：

public?class?TaggedWritable?extends?TaggedMapOutput?{private?Text?data;public?TaggedWritable()?{this.tag?=?new?Text("");this.data?=?new?Text("");}public?TaggedWritable(Text?data)?{this.data?=?data;}@Overridepublic?void?readFields(DataInput?in)?throws?IOException?{this.tag.readFields(in);this.data.readFields(in);}@Overridepublic?void?write(DataOutput?out)?throws?IOException?{this.tag.write(out);this.data.write(out);}@Overridepublic?Text?getData()?{return?data;} }

第二部分，map函數

public?class?Mapclass?extends?DataJoinMapperBase{@Overrideprotected?Text?generateGroupKey(TaggedMapOutput?aRecord)?{String?line?=?aRecord.getData().toString();String[]?tokens?=?line.split(",");String?groupKey?=?tokens[0];return?new?Text(groupKey);}@Overrideprotected?Text?generateInputTag(String?inputFile)?{String?datasource?=?inputFile.split("-")[0];return?new?Text(datasource);}@Overrideprotected?TaggedWritable?generateTaggedMapOutput(Object?value)?{TaggedWritable?retv?=?new?TaggedWritable(new?Text(value.toString()));retv.setTag(this.inputTag);return?retv;} }

map函數中要特別注意protected TaggedWritable generateTaggedMapOutput(Object value) 該方法的返回類型為你第一步定義的類型。

第三部分，reduce

public?class?Reduce?extends?DataJoinReducerBase{@Overrideprotected?TaggedMapOutput?combine(Object[]?tags,?Object[]?values)?{if?(tags.length?<?2)?{return?null;????}String?joinedStr?=?"";???for?(int?i=0;?i<values.length;?i++)?{??if?(i?>?0){joinedStr?+=?",";?}TaggedWritable?tw?=?(TaggedWritable)?values[i];??String?line?=?tw.getData().toString();??System.out.println("line=========:"+line);String[]?tokens?=?line.split(",",?2);??joinedStr?+=?tokens[1];??}??TaggedWritable?retv?=?new?TaggedWritable(new?Text(joinedStr));??retv.setTag((Text)?tags[0]);???return?retv;??} }

reduce過程會將主鍵與數據組合在一起輸出，你拼接的字符串中無需寫入主鍵。

public?class?Datajoin?extends?Configured?implements?Tool?{@Overridepublic?int?run(String[]?args)?throws?Exception?{Configuration?conf?=?this.getConf();JobConf?job?=?new?JobConf(conf,?Datajoin.class);job.setJarByClass(Datajoin.class);Path?in?=?new?Path("hdfs://172.16.0.87:9000/user/jeff/datajoin/");Path?out?=?new?Path("hdfs://172.16.0.87:9000/user/jeff/datajoin/out");FileInputFormat.setInputPaths(job,?in);FileOutputFormat.setOutputPath(job,?out);job.setJobName("DataJoin");job.setMapperClass(Mapclass.class);job.setReducerClass(Reduce.class);job.setInputFormat(TextInputFormat.class);job.setOutputFormat(TextOutputFormat.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(TaggedWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);job.set("mapred.textoutputformat.separator",?",");JobClient.runJob(job);return?0;}public?static?void?main(String[]?args)?throws?Exception?{int?res?=?ToolRunner.run(new?Configuration(),?new?Datajoin(),?args);System.exit(res);} }

運行mapreduce任務后的輸出為：

1,Stephanie Leung,555-555-5555,B,88.25,20-May-2008
2,Edward Kim,123-456-7890,C,32.00,30-Nov-2007
3,Jose Madriz,281-330-8004,A,12.95,02-Jun-2008
3,Jose Madriz,281-330-8004,D,25.02,22-Jan-2009

可以在reduce的combin函數中控制函數的輸出形式，左聯，或者右聯。

轉載于:https://my.oschina.net/hanjiafu/blog/291614

總結

以上是生活随笔為你收集整理的hadoop datajoin的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。