Hadoop 之 MapReduce 的工作原理及其倒排索引的建立
一、Hadoop?簡介
?
下面先從一張圖理解MapReduce得整個(gè)工作原理
下面對上面出現(xiàn)的一些名詞進(jìn)行介紹
ResourceManager：是YARN資源控制框架的中心模塊，負(fù)責(zé)集群中所有的資源的統(tǒng)一管理和分配。它接收來自NM(NodeManager)的匯報(bào)，建立AM，并將資源派送給AM(ApplicationMaster)。
NodeManager:簡稱NM，NodeManager是ResourceManager在每臺機(jī)器的上代理，負(fù)責(zé)容器的管理，并監(jiān)控他們的資源使用情況（cpu，內(nèi)存，磁盤及網(wǎng)絡(luò)等），以及向 ResourceManager提供這些資源使用報(bào)告。
ApplicationMaster:以下簡稱AM。YARN中每個(gè)應(yīng)用都會啟動一個(gè)AM，負(fù)責(zé)向RM申請資源，請求NM啟動container，并告訴container做什么事情。
Container：資源容器。YARN中所有的應(yīng)用都是在container之上運(yùn)行的。AM也是在container上運(yùn)行的，不過AM的container是RM申請的。
?
1.? ? Container是YARN中資源的抽象，它封裝了某個(gè)節(jié)點(diǎn)上一定量的資源（CPU和內(nèi)存兩類資源）。
2.? Container由ApplicationMaster向ResourceManager申請的，由ResouceManager中的資源調(diào)度器異步分配給ApplicationMaster；
3.? ? Container的運(yùn)行是由ApplicationMaster向資源所在的NodeManager發(fā)起的，Container運(yùn)行時(shí)需提供內(nèi)部執(zhí)行的任務(wù)命令（可以是任何命令，比如java、Python、C++進(jìn)程啟動命令均可）以及該命令執(zhí)行所需的環(huán)境變量和外部資源（比如詞典文件、可執(zhí)行文件、jar包等）。
另外，一個(gè)應(yīng)用程序所需的Container分為兩大類，如下：
? ? ? ? ? （1） 運(yùn)行ApplicationMaster的Container：這是由ResourceManager（向內(nèi)部的資源調(diào)度器）申請和啟動的，用戶提交應(yīng)用程序時(shí)，可指定唯一的ApplicationMaster所需的資源；
?????? （2） 運(yùn)行各類任務(wù)的Container：這是由ApplicationMaster向ResourceManager申請的，并由ApplicationMaster與NodeManager通信以啟動之。
以上兩類Container可能在任意節(jié)點(diǎn)上，它們的位置通常而言是隨機(jī)的，即ApplicationMaster可能與它管理的任務(wù)運(yùn)行在一個(gè)節(jié)點(diǎn)上。
整個(gè)MapReduce的過程大致分為 Map-->Shuffle（排序）-->Combine（組合）-->Reduce
下面通過一個(gè)單詞計(jì)數(shù)案例來理解各個(gè)過程
1）將文件拆分成splits(片)，并將每個(gè)split按行分割形成<key,value>對，如圖所示。這一步由MapReduce框架自動完成，其中偏移量即key值
?? ??? ??? ??? ??? ?
?? ??? ??? ??? ??? ?分割過程
將分割好的<key,value>對交給用戶定義的map方法進(jìn)行處理，生成新的<key,value>對，如下圖所示。
?? ??? ??? ??? ??? ?執(zhí)行map方法
?? ??? ??? ??? ??? ?
得到map方法輸出的<key,value>對后，Mapper會將它們按照key值進(jìn)行Shuffle（排序），并執(zhí)行Combine過程，將key至相同value值累加，得到Mapper的最終輸出結(jié)果。如下圖所示。
?? ??? ??? ??? ??? ? Map端排序及Combine過程
Reducer先對從Mapper接收的數(shù)據(jù)進(jìn)行排序，再交由用戶自定義的reduce方法進(jìn)行處理，得到新的<key,value>對，并作為WordCount的輸出結(jié)果，如下圖所示。
?? ??? ??? ??? ??? ?Reduce端排序及輸出結(jié)果
下面看怎么用Java來實(shí)現(xiàn)WordCount單詞計(jì)數(shù)的功能
首先看Map過程
Map過程需要繼承org.apache.hadoop.mapreduce.Mapper包中 Mapper 類，并重寫其map方法。
?
?
二、通過 Hadoop 建立倒排索引
倒排索引就是根據(jù)單詞內(nèi)容來查找文檔的方式，由于不是根據(jù)文檔來確定文檔所包含的內(nèi)容，進(jìn)行了相反的操作，所以被稱為倒排索引,?它是搜索引擎最為核心的數(shù)據(jù)結(jié)構(gòu)，以及文檔檢索的關(guān)鍵部分。
下面來看一個(gè)例子來理解什么是倒排索引
這里我準(zhǔn)備了兩個(gè)文件 分別為1.txt和2.txt
1.txt的內(nèi)容如下
I Love HadoopI like ZhouSiYuanI love me?
2.txt的內(nèi)容如下
I Love MapReduce I like NBA I love Hadoop?
我這里使用的是默認(rèn)的輸入格式TextInputFormat，他是一行一行的讀的，鍵是偏移量。
?
所以在map階段之前的到結(jié)果如下?
map階段從1.txt的得到的輸入
map階段從2.txt的得到的輸入
0 I Love MapReduce 18 I like NBA 30 I love Hadoopmap階段?
把詞頻作為值?
把單詞和URI組成key值?
比如?
key : I+hdfs://192.168.52.140:9000/index/2.txt value:1
為什么要這樣設(shè)置鍵和值？?
因?yàn)檫@樣設(shè)計(jì)可以使用MapReduce框架自帶的map端排序，將同一單詞的詞頻組成列表
經(jīng)過map階段1.txt得到的輸出如下
I:hdfs://192.168.52.140:9000/index/1.txt 1 Love:hdfs://192.168.52.140:9000/index/1.txt 1 MapReduce:hdfs://192.168.52.140:9000/index/1.txt 1 I:hdfs://192.168.52.140:9000/index/1.txt 1 Like:hdfs://192.168.52.140:9000/index/1.txt 1 ZhouSiYuan:hdfs://192.168.52.140:9000/index/1.txt 1 I:hdfs://192.168.52.140:9000/index/1.txt 1 love:hdfs://192.168.52.140:9000/index/1.txt 1 me:hdfs://192.168.52.140:9000/index/1.txt 1經(jīng)過map階段2.txt得到的輸出如下
I:hdfs://192.168.52.140:9000/index/2.txt 1 Love:hdfs://192.168.52.140:9000/index/2.txt 1 MapReduce:hdfs://192.168.52.140:9000/index/2.txt 1 I:hdfs://192.168.52.140:9000/index/2.txt 1 Like:hdfs://192.168.52.140:9000/index/2.txt 1 NBA:hdfs://192.168.52.140:9000/index/2.txt 1 I:hdfs://192.168.52.140:9000/index/2.txt 1 love:hdfs://192.168.52.140:9000/index/2.txt 1 Hadoop:hdfs://192.168.52.140:9000/index/2.txt 11.txt經(jīng)過MapReduce框架自帶的map端排序得到的輸出結(jié)果如下
I:hdfs://192.168.52.140:9000/index/1.txt list{1,1,1} Love:hdfs://192.168.52.140:9000/index/1.txt list{1} MapReduce:hdfs://192.168.52.140:9000/index/1.txt list{1} Like:hdfs://192.168.52.140:9000/index/1.txt list{1} ZhouSiYuan:hdfs://192.168.52.140:9000/index/1.txt list{1} love:hdfs://192.168.52.140:9000/index/1.txt list{1} me:hdfs://192.168.52.140:9000/index/1.txt list{1}2.txt經(jīng)過MapReduce框架自帶的map端排序得到的輸出結(jié)果如下
?
I:hdfs://192.168.52.140:9000/index/2.txt list{1,1,1} Love:hdfs://192.168.52.140:9000/index/2.txt list{1} MapReduce:hdfs://192.168.52.140:9000/index/2.txt list{1} Like:hdfs://192.168.52.140:9000/index/2.txt list{1} NBA:hdfs://192.168.52.140:9000/index/2.txt list{1} love:hdfs://192.168.52.140:9000/index/2.txt list{1} Hadoop:hdfs://192.168.52.140:9000/index/2.txt list{1}combine階段：?
key值為單詞，?
value值由URI和詞頻組成?
value: hdfs://192.168.52.140:9000/index/2.txt:3 key:I?
為什么這樣設(shè)計(jì)鍵值了？?
因?yàn)樵赟huffle過程將面臨一個(gè)問題，所有具有相同單詞的記錄(由單詞、URL和詞頻組成)應(yīng)該交由同一個(gè)Reducer處理?
所以重新把單詞設(shè)置為鍵可以使用MapReduce框架默認(rèn)的Shuffle過程,將相同單詞的所有記錄發(fā)送給同一個(gè)Reducer處理
combine階段將key相同的value值累加
1.txt得到如下輸出
I hdfs://192.168.52.140:9000/index/1.txt:3 Love hdfs://192.168.52.140:9000/index/1.txt:1 MapReduce hdfs://192.168.52.140:9000/index/1.txt:1 Like hdfs://192.168.52.140:9000/index/1.txt:1 ZhouSiYuan hdfs://192.168.52.140:9000/index/1.txt:1 love hdfs://192.168.52.140:9000/index/1.txt:1 me hdfs://192.168.52.140:9000/index/1.txt:12.txt得到如下輸出
I hdfs://192.168.52.140:9000/index/2.txt:3 Love hdfs://192.168.52.140:9000/index/2.txt:1 MapReduce hdfs://192.168.52.140:9000/index/2.txt:1 Like hdfs://192.168.52.140:9000/index/2.txt:1 NBA hdfs://192.168.52.140:9000/index/2.txt:1 love hdfs://192.168.52.140:9000/index/2.txt:1 Hadoop hdfs://192.168.52.140:9000/index/2.txt:1這樣reducer過程就很簡單了，它只用來生成文檔列表?
比如相同的單詞I，這樣生成文檔列表?
I hdfs://192.168.52.140:9000/index/2.txt:3;hdfs://192.168.52.140:9000/index/1.txt:3;
最后所有的輸出結(jié)果如下
Hadoop hdfs://192.168.52.140:9000/index/1.txt:1;hdfs://192.168.52.140:9000/index/2.txt:1; I hdfs://192.168.52.140:9000/index/2.txt:3;hdfs://192.168.52.140:9000/index/1.txt:3; Love hdfs://192.168.52.140:9000/index/1.txt:1;hdfs://192.168.52.140:9000/index/2.txt:1; MapReduce hdfs://192.168.52.140:9000/index/2.txt:1; NBA hdfs://192.168.52.140:9000/index/2.txt:1; ZhouSiYuan hdfs://192.168.52.140:9000/index/1.txt:1; like hdfs://192.168.52.140:9000/index/1.txt:1;hdfs://192.168.52.140:9000/index/2.txt:1; love hdfs://192.168.52.140:9000/index/2.txt:1;hdfs://192.168.52.140:9000/index/1.txt:1; me hdfs://192.168.52.140:9000/index/1.txt:1;下面是整個(gè)源代碼
package com.hadoop.mapreduce.test8.invertedindex;import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.FileSplit; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class InvertedIndex {/*** * @author 湯高**/public static class InvertedIndexMapper extends Mapper<Object, Text, Text, Text>{private Text keyInfo = new Text(); // 存儲單詞和URI的組合private Text valueInfo = new Text(); //存儲詞頻private FileSplit split; // 存儲split對象。@Overrideprotected void map(Object key, Text value, Mapper<Object, Text, Text, Text>.Context context)throws IOException, InterruptedException {//獲得<key,value>對所屬的FileSplit對象。split = (FileSplit) context.getInputSplit();System.out.println("偏移量"+key);System.out.println("值"+value);//StringTokenizer是用來把字符串截取成一個(gè)個(gè)標(biāo)記或單詞的，默認(rèn)是空格或多個(gè)空格(\t\n\r等等)截取StringTokenizer itr = new StringTokenizer( value.toString());while( itr.hasMoreTokens() ){// key值由單詞和URI組成。keyInfo.set( itr.nextToken()+":"+split.getPath().toString());//詞頻初始為1valueInfo.set("1");context.write(keyInfo, valueInfo);}System.out.println("key"+keyInfo);System.out.println("value"+valueInfo);}}/*** * @author 湯高**/public static class InvertedIndexCombiner extends Reducer<Text, Text, Text, Text>{private Text info = new Text();@Overrideprotected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)throws IOException, InterruptedException {//統(tǒng)計(jì)詞頻int sum = 0;for (Text value : values) {sum += Integer.parseInt(value.toString() );}int splitIndex = key.toString().indexOf(":");//重新設(shè)置value值由URI和詞頻組成info.set( key.toString().substring( splitIndex + 1) +":"+sum );//重新設(shè)置key值為單詞key.set( key.toString().substring(0,splitIndex));context.write(key, info);System.out.println("key"+key);System.out.println("value"+info);}}/*** * @author 湯高**/public static class InvertedIndexReducer extends Reducer<Text, Text, Text, Text>{private Text result = new Text();@Overrideprotected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)throws IOException, InterruptedException {//生成文檔列表String fileList = new String();for (Text value : values) {fileList += value.toString()+";";}result.set(fileList);context.write(key, result);}}public static void main(String[] args) {try {Configuration conf = new Configuration();Job job = Job.getInstance(conf,"InvertedIndex");job.setJarByClass(InvertedIndex.class);//實(shí)現(xiàn)map函數(shù)，根據(jù)輸入的<key,value>對生成中間結(jié)果。job.setMapperClass(InvertedIndexMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Text.class);job.setCombinerClass(InvertedIndexCombiner.class);job.setReducerClass(InvertedIndexReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);//我把那兩個(gè)文件上傳到這個(gè)index目錄下了FileInputFormat.addInputPath(job, new Path("hdfs://192.168.52.140:9000/index/"));//把結(jié)果輸出到out_index+時(shí)間戳的目錄下FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.52.140:9000/out_index"+System.currentTimeMillis()+"/"));System.exit(job.waitForCompletion(true) ? 0 : 1);} catch (IllegalStateException e) {e.printStackTrace();} catch (IllegalArgumentException e) {e.printStackTrace();} catch (ClassNotFoundException e) {e.printStackTrace();} catch (IOException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}} }轉(zhuǎn)載自：? ?https://blog.csdn.net/tanggao1314/article/details/51340672
總結(jié)
以上是生活随笔為你收集整理的Hadoop 之 MapReduce 的工作原理及其倒排索引的建立的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 利用C++实现自定义插件
- 下一篇： MicroSoft CryptoAPI