mapreduce文本排序

自上次發(fā)布以來已經(jīng)有一段時(shí)間了，因?yàn)槲乙恢泵τ贑oursera提供的一些課程。 有一些非常有趣的產(chǎn)品，值得一看。 前一段時(shí)間，我購買了Jimmy Lin和Chris Dyer的MapReduce數(shù)據(jù)密集型處理程序 。 本書以偽代碼格式介紹了幾種關(guān)鍵的MapReduce算法。 我的目標(biāo)是采用第3-6章中介紹的算法，并以Hadoop： Tom White的《權(quán)威指南》作為參考在Hadoop中實(shí)現(xiàn)它們。 我將假設(shè)您熟悉Hadoop和MapReduce，并且不介紹任何入門資料。 因此，讓我們從局部聚合開始進(jìn)入第3章– MapReduce算法設(shè)計(jì)。

本地聚集

在很高的級別上，當(dāng)Mappers發(fā)出數(shù)據(jù)時(shí)，中間結(jié)果將寫入磁盤，然后通過網(wǎng)絡(luò)發(fā)送到Reducers以進(jìn)行最終處理。 在MapReduce作業(yè)的處理中，寫入磁盤然后通過網(wǎng)絡(luò)傳輸數(shù)據(jù)的延遲是一項(xiàng)昂貴的操作。 因此，有理由認(rèn)為，只要有可能，減少從映射器發(fā)送的數(shù)據(jù)量將提高M(jìn)apReduce作業(yè)的速度。 本地聚合是一種用于減少數(shù)據(jù)量并提高M(jìn)apReduce工作效率的技術(shù)。 局部聚合不能代替reducers，因?yàn)槲覀冃枰环N方法來使用來自不同映射器的相同鍵來收集結(jié)果。 我們將考慮實(shí)現(xiàn)本地聚合的3種方法：

使用Hadoop Combiner函數(shù)。

文本處理和MapReduce書中介紹了兩種“內(nèi)部映射器”組合方法。

當(dāng)然，任何優(yōu)化都需要權(quán)衡取舍，我們也會對此進(jìn)行討論。
為了演示本地聚合，我們將使用hadoop-0.20.2-cdh3u3在MacBookPro上安裝的偽分布式集群上，在Charles Dickens的純文本版本的A Christmas Carol （從Project Gutenberg下載）上運(yùn)行無處不在的字?jǐn)?shù)統(tǒng)計(jì)作業(yè)。從Cloudera發(fā)行。 我計(jì)劃在以后的文章中在具有更實(shí)際大小的數(shù)據(jù)的EC2集群上運(yùn)行相同的實(shí)驗(yàn)。

合路器

組合器函數(shù)是擴(kuò)展Reducer類的對象。 實(shí)際上，對于此處的示例，我們將重復(fù)使用單詞計(jì)數(shù)作業(yè)中使用的相同的reduce。 設(shè)置MapReduce作業(yè)時(shí)會指定組合器功能，如下所示：

job.setReducerClass(TokenCountReducer.class);

這是化簡器代碼：

public class TokenCountReducer extends Reducer<Text,IntWritable,Text,IntWritable>{@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int count = 0;for (IntWritable value : values) {count+= value.get();}context.write(key,new IntWritable(count));} }

組合器的工作就是按照名稱所暗示的那樣進(jìn)行工作，聚合數(shù)據(jù)的最終結(jié)果是減少了數(shù)據(jù)在網(wǎng)絡(luò)中的混洗，從而提高了效率。 如前所述，請記住，還需要簡化器將結(jié)果與來自不同映射器的相同鍵組合在一起。 由于組合器功能是一種優(yōu)化，因此Hadoop框架無法保證組合器將被調(diào)用多少次（如果有的話）。

在Mapper組合選項(xiàng)1中

使用組合器的第一種選擇（圖3.2，第41頁）非常簡單，對我們原始的字?jǐn)?shù)映射器進(jìn)行了一些修改：

public class PerDocumentMapper extends Mapper<LongWritable, Text, Text, IntWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {IntWritable writableCount = new IntWritable();Text text = new Text();Map<String,Integer> tokenMap = new HashMap<String, Integer>();StringTokenizer tokenizer = new StringTokenizer(value.toString());while(tokenizer.hasMoreElements()){String token = tokenizer.nextToken();Integer count = tokenMap.get(token);if(count == null) count = new Integer(0);count+=1;tokenMap.put(token,count);}Set<String> keys = tokenMap.keySet();for (String s : keys) {text.set(s);writableCount.set(tokenMap.get(s));context.write(text,writableCount);}} }

正如我們在這里看到的，對于遇到的每個(gè)單詞，我們不會發(fā)出計(jì)數(shù)為1的單詞，而是使用映射來跟蹤已處理的每個(gè)單詞。 然后，在處理完所有標(biāo)記后，我們遍歷該映射并發(fā)出該行中遇到的每個(gè)單詞的總數(shù)。

在Mapper組合選項(xiàng)2中

in mapper組合的第二個(gè)選項(xiàng)（圖3.3，第41頁）與上述示例非常相似，但有兩個(gè)區(qū)別–創(chuàng)建哈希映射時(shí)和發(fā)出映射中包含的結(jié)果時(shí)。 在上面的示例中，創(chuàng)建了一個(gè)映射，并在每次調(diào)用map方法時(shí)將其內(nèi)容轉(zhuǎn)儲到網(wǎng)絡(luò)上。 在此示例中，我們將使地圖成為實(shí)例變量，并將地圖的實(shí)例化移動(dòng)到我們的映射器中的setUp方法。 同樣，在完成對mapper的所有調(diào)用并調(diào)用cleanUp方法之前，不會將映射的內(nèi)容發(fā)送到reducers。

public class AllDocumentMapper extends Mapper<LongWritable,Text,Text,IntWritable> {private Map<String,Integer> tokenMap;@Overrideprotected void setup(Context context) throws IOException, InterruptedException {tokenMap = new HashMap<String, Integer>();}@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {StringTokenizer tokenizer = new StringTokenizer(value.toString());while(tokenizer.hasMoreElements()){String token = tokenizer.nextToken();Integer count = tokenMap.get(token);if(count == null) count = new Integer(0);count+=1;tokenMap.put(token,count);}}@Overrideprotected void cleanup(Context context) throws IOException, InterruptedException {IntWritable writableCount = new IntWritable();Text text = new Text();Set<String> keys = tokenMap.keySet();for (String s : keys) {text.set(s);writableCount.set(tokenMap.get(s));context.write(text,writableCount);}} }

從上面的代碼示例中可以看出，在對map方法的所有調(diào)用中，映射器都跟蹤唯一字?jǐn)?shù)。 通過跟蹤唯一令牌及其計(jì)數(shù)，應(yīng)該大大減少發(fā)送給reducer的記錄數(shù)量，這反過來又可以改善MapReduce作業(yè)的運(yùn)行時(shí)間。 這樣可以達(dá)到與使用MapReduce框架提供的Combiner Function選項(xiàng)相同的效果，但是在這種情況下，可以確保將調(diào)用組合代碼。 但是這種方法也有一些警告。 在地圖調(diào)用之間保持狀態(tài)可能會出現(xiàn)問題，并且絕對違反了“地圖”功能的功能精神。 同樣，通過保持所有映射器的狀態(tài)，取決于作業(yè)中使用的數(shù)據(jù)，內(nèi)存可能是另一個(gè)要解決的問題。 最終，必須權(quán)衡所有權(quán)衡以確定最佳方法。

結(jié)果

現(xiàn)在，讓我們看一下不同映射器的一些結(jié)果。 由于作業(yè)是在偽分布式模式下運(yùn)行的，因此實(shí)際的運(yùn)行時(shí)間是無關(guān)緊要的，但是我們?nèi)匀豢梢酝茢喑鍪褂帽镜鼐酆蠒绾斡绊懺趯?shí)際集群上運(yùn)行的MapReduce作業(yè)的效率。

每個(gè)令牌映射器：

12/09/13 21:25:32 INFO mapred.JobClient: Reduce shuffle bytes=366010 12/09/13 21:25:32 INFO mapred.JobClient: Reduce output records=7657 12/09/13 21:25:32 INFO mapred.JobClient: Spilled Records=63118 12/09/13 21:25:32 INFO mapred.JobClient: Map output bytes=302886

在Mapper精簡選項(xiàng)1中：

12/09/13 21:28:15 INFO mapred.JobClient: Reduce shuffle bytes=354112 12/09/13 21:28:15 INFO mapred.JobClient: Reduce output records=7657 12/09/13 21:28:15 INFO mapred.JobClient: Spilled Records=60704 12/09/13 21:28:15 INFO mapred.JobClient: Map output bytes=293402

在Mapper精簡選項(xiàng)2中：

12/09/13 21:30:49 INFO mapred.JobClient: Reduce shuffle bytes=105885 12/09/13 21:30:49 INFO mapred.JobClient: Reduce output records=7657 12/09/13 21:30:49 INFO mapred.JobClient: Spilled Records=15314 12/09/13 21:30:49 INFO mapred.JobClient: Map output bytes=90565

組合器選項(xiàng)：

12/09/13 21:22:18 INFO mapred.JobClient: Reduce shuffle bytes=105885 12/09/13 21:22:18 INFO mapred.JobClient: Reduce output records=7657 12/09/13 21:22:18 INFO mapred.JobClient: Spilled Records=15314 12/09/13 21:22:18 INFO mapred.JobClient: Map output bytes=302886 12/09/13 21:22:18 INFO mapred.JobClient: Combine input records=31559 12/09/13 21:22:18 INFO mapred.JobClient: Combine output records=7657

不出所料，不進(jìn)行合并的Mapper的結(jié)果最差，緊隨其后的是第一個(gè)映射器內(nèi)合并選項(xiàng)（盡管如果在運(yùn)行字?jǐn)?shù)統(tǒng)計(jì)之前清除數(shù)據(jù)，這些結(jié)果本來可以做得更好）。 第二個(gè)映射器內(nèi)合并選項(xiàng)和合并器功能實(shí)際上具有相同的結(jié)果。 重要的事實(shí)是，作為前兩個(gè)選項(xiàng)，兩者產(chǎn)生的結(jié)果都減少了2/3，減少了混洗字節(jié)。 將通過網(wǎng)絡(luò)發(fā)送到縮減程序的字節(jié)數(shù)量減少該數(shù)量一定會對MapReduce作業(yè)的效率產(chǎn)生積極影響。 這里要牢記一點(diǎn)，那就是合并器/映射器合并不能僅在所有MapReduce作業(yè)中使用，在這種情況下，字?jǐn)?shù)計(jì)數(shù)非常適合于這種增強(qiáng)，但這可能并不總是正確的。

結(jié)論

正如您所看到的，在尋求提高M(jìn)apReduce作業(yè)的性能時(shí)，需要認(rèn)真考慮使用映射器內(nèi)合并或Hadoop合并器功能的好處。 至于哪種方法，則要權(quán)衡每種方法的權(quán)衡。

相關(guān)鏈接

• Jimmy Lin和Chris Dyer 使用MapReduce進(jìn)行的數(shù)據(jù)密集型處理
• Hadoop： Tom White 的權(quán)威指南
• 來自博客的源代碼
• MRUnit用于單元測試Apache Hadoop映射減少工作
• Gutenberg項(xiàng)目提供了大量純文本格式的書籍，非常適合在本地測試Hadoop作業(yè)。

祝您編程愉快，別忘了分享！

參考： 《 隨機(jī)編碼》博客上的JCG合作伙伴 Bill Bejeck 提供的MapReduce數(shù)據(jù)密集型文本處理功能 。

翻譯自: https://www.javacodegeeks.com/2012/09/mapreduce-working-through-data.html

mapreduce文本排序

總結(jié)

以上是生活随笔為你收集整理的mapreduce文本排序_MapReduce：通过数据密集型文本处理的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。