一、Hadoop?簡介

?

下面先從一張圖理解MapReduce得整個工作原理



下面對上面出現的一些名詞進行介紹


ResourceManager：是YARN資源控制框架的中心模塊，負責集群中所有的資源的統一管理和分配。它接收來自NM(NodeManager)的匯報，建立AM，并將資源派送給AM(ApplicationMaster)。

NodeManager:簡稱NM，NodeManager是ResourceManager在每臺機器的上代理，負責容器的管理，并監控他們的資源使用情況（cpu，內存，磁盤及網絡等），以及向 ResourceManager提供這些資源使用報告。

ApplicationMaster:以下簡稱AM。YARN中每個應用都會啟動一個AM，負責向RM申請資源，請求NM啟動container，并告訴container做什么事情。

Container：資源容器。YARN中所有的應用都是在container之上運行的。AM也是在container上運行的，不過AM的container是RM申請的。

?

1.? ? Container是YARN中資源的抽象，它封裝了某個節點上一定量的資源（CPU和內存兩類資源）。

2.? Container由ApplicationMaster向ResourceManager申請的，由ResouceManager中的資源調度器異步分配給ApplicationMaster；
3.? ? Container的運行是由ApplicationMaster向資源所在的NodeManager發起的，Container運行時需提供內部執行的任務命令（可以是任何命令，比如java、Python、C++進程啟動命令均可）以及該命令執行所需的環境變量和外部資源（比如詞典文件、可執行文件、jar包等）。
另外，一個應用程序所需的Container分為兩大類，如下：
? ? ? ? ? （1） 運行ApplicationMaster的Container：這是由ResourceManager（向內部的資源調度器）申請和啟動的，用戶提交應用程序時，可指定唯一的ApplicationMaster所需的資源；
?????? （2） 運行各類任務的Container：這是由ApplicationMaster向ResourceManager申請的，并由ApplicationMaster與NodeManager通信以啟動之。
以上兩類Container可能在任意節點上，它們的位置通常而言是隨機的，即ApplicationMaster可能與它管理的任務運行在一個節點上。


整個MapReduce的過程大致分為 Map-->Shuffle（排序）-->Combine（組合）-->Reduce





下面通過一個單詞計數案例來理解各個過程
1）將文件拆分成splits(片)，并將每個split按行分割形成<key,value>對，如圖所示。這一步由MapReduce框架自動完成，其中偏移量即key值



?? ??? ??? ??? ??? ?
?? ??? ??? ??? ??? ?分割過程



將分割好的<key,value>對交給用戶定義的map方法進行處理，生成新的<key,value>對，如下圖所示。


?? ??? ??? ??? ??? ?執行map方法
?? ??? ??? ??? ??? ?



得到map方法輸出的<key,value>對后，Mapper會將它們按照key值進行Shuffle（排序），并執行Combine過程，將key至相同value值累加，得到Mapper的最終輸出結果。如下圖所示。

?? ??? ??? ??? ??? ? Map端排序及Combine過程



Reducer先對從Mapper接收的數據進行排序，再交由用戶自定義的reduce方法進行處理，得到新的<key,value>對，并作為WordCount的輸出結果，如下圖所示。


?? ??? ??? ??? ??? ?Reduce端排序及輸出結果





下面看怎么用Java來實現WordCount單詞計數的功能

首先看Map過程
Map過程需要繼承org.apache.hadoop.mapreduce.Mapper包中 Mapper 類，并重寫其map方法。

?

/*** Mapper<LongWritable, Text, Text, IntWritable>中 LongWritable,IntWritable是Hadoop數據類型表示長整型和整形** LongWritable, Text表示輸入類型 (比如本應用單詞計數輸入是 偏移量(字符串中的第一個單詞的其實位置),對應的單詞(值))* Text, IntWritable表示輸出類型 輸出是單詞 和他的個數* 注意：map函數中前兩個參數LongWritable key, Text value和輸出類型不一致* 所以后面要設置輸出類型 要使他們一致*///Map過程public static class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {/*****/@Overrideprotected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)throws IOException, InterruptedException {//默認的map的value是每一行,我這里自定義的是以空格分割String[] vs = value.toString().split("\\s");for (String v : vs) {//寫出去context.write(new Text(v), ONE);}}} Reduce過程 Reduce過程需要繼承org.apache.hadoop.mapreduce包中 Reducer 類，并 重寫 其reduce方法。Map過程輸出<key,values>中key為單個單詞，而values是對應單詞的計數值所組成的列表，Map的輸出就是Reduce的輸入，所以reduce方法只要遍歷values并求和，即可得到某個單詞的總次數。 //Reduce過程/**** Text, IntWritable輸入類型,從map過程獲得 既map的輸出作為Reduce的輸入* Text, IntWritable輸出類型*/public static class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{@Overrideprotected void reduce(Text key, Iterable<IntWritable> values,Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {int count=0;for(IntWritable v:values){count+=v.get();//單詞個數加一}context.write(key, new IntWritable(count));}} 最后執行MapReduce任務 public static void main(String[] args) {Configuration conf=new Configuration();try {//args從控制臺獲取路徑 解析得到域名String[] paths=new GenericOptionsParser(conf,args).getRemainingArgs();if(paths.length<2){throw new RuntimeException("必須輸出 輸入 和輸出路徑");}//得到一個Job 并設置名字Job job=Job.getInstance(conf,"wordcount");//設置Jar 使本程序在Hadoop中運行job.setJarByClass(WordCount.class);//設置Map處理類job.setMapperClass(WordCountMapper.class);//設置map的輸出類型,因為不一致,所以要設置job.setMapOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);//設置Reduce處理類job.setReducerClass(WordCountReducer.class);//設置輸入和輸出目錄FileInputFormat.addInputPath(job, new Path(paths[0]));FileOutputFormat.setOutputPath(job, new Path(paths[1]));//啟動運行System.exit(job.waitForCompletion(true) ? 0:1);} catch (IOException e) {e.printStackTrace();} catch (ClassNotFoundException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}} 即可求得每個單詞的個數下面把整個過程的源碼附上,有需要的朋友可以拿去測試 package hadoopday02;import java.io.IOException;import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable;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.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser;public class WordCount {//計數變量private static final IntWritable ONE = new IntWritable(1);/**** @author 湯高* Mapper<LongWritable, Text, Text, IntWritable>中 LongWritable,IntWritable是Hadoop數據類型表示長整型和整形** LongWritable, Text表示輸入類型 (比如本應用單詞計數輸入是 偏移量(字符串中的第一個單詞的其實位置),對應的單詞(值))* Text, IntWritable表示輸出類型 輸出是單詞 和他的個數* 注意：map函數中前兩個參數LongWritable key, Text value和輸出類型不一致* 所以后面要設置輸出類型 要使他們一致*///Map過程public static class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {/*****/@Overrideprotected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)throws IOException, InterruptedException {//默認的map的value是每一行,我這里自定義的是以空格分割String[] vs = value.toString().split("\\s");for (String v : vs) {//寫出去context.write(new Text(v), ONE);}}}//Reduce過程/**** Text, IntWritable輸入類型,從map過程獲得 既map的輸出作為Reduce的輸入* Text, IntWritable輸出類型*/public static class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{@Overrideprotected void reduce(Text key, Iterable<IntWritable> values,Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {int count=0;for(IntWritable v:values){count+=v.get();//單詞個數加一}context.write(key, new IntWritable(count));}}public static void main(String[] args) {Configuration conf=new Configuration();try {//args從控制臺獲取路徑 解析得到域名String[] paths=new GenericOptionsParser(conf,args).getRemainingArgs();if(paths.length<2){throw new RuntimeException("必須輸出 輸入 和輸出路徑");}//得到一個Job 并設置名字Job job=Job.getInstance(conf,"wordcount");//設置Jar 使本程序在Hadoop中運行job.setJarByClass(WordCount.class);//設置Map處理類job.setMapperClass(WordCountMapper.class);//設置map的輸出類型,因為不一致,所以要設置job.setMapOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);//設置Reduce處理類job.setReducerClass(WordCountReducer.class);//設置輸入和輸出目錄FileInputFormat.addInputPath(job, new Path(paths[0]));FileOutputFormat.setOutputPath(job, new Path(paths[1]));//啟動運行System.exit(job.waitForCompletion(true) ? 0:1);} catch (IOException e) {e.printStackTrace();} catch (ClassNotFoundException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}} }

?

二、通過 Hadoop 建立倒排索引

倒排索引就是根據單詞內容來查找文檔的方式，由于不是根據文檔來確定文檔所包含的內容，進行了相反的操作，所以被稱為倒排索引,?它是搜索引擎最為核心的數據結構，以及文檔檢索的關鍵部分。

下面來看一個例子來理解什么是倒排索引

這里我準備了兩個文件 分別為1.txt和2.txt

1.txt的內容如下

I Love HadoopI like ZhouSiYuanI love me

---

?

2.txt的內容如下

I Love MapReduce I like NBA I love Hadoop

?

我這里使用的是默認的輸入格式TextInputFormat，他是一行一行的讀的，鍵是偏移量。
?

所以在map階段之前的到結果如下?
map階段從1.txt的得到的輸入

0 I Love Hadoop 15 I like ZhouSiYuan 34 I love me

map階段從2.txt的得到的輸入

0 I Love MapReduce 18 I like NBA 30 I love Hadoop

map階段?
把詞頻作為值?
把單詞和URI組成key值?
比如?
key : I+hdfs://192.168.52.140:9000/index/2.txt value:1

為什么要這樣設置鍵和值？?
因為這樣設計可以使用MapReduce框架自帶的map端排序，將同一單詞的詞頻組成列表

經過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

經過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 1

1.txt經過MapReduce框架自帶的map端排序得到的輸出結果如下

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經過MapReduce框架自帶的map端排序得到的輸出結果如下

?

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?
為什么這樣設計鍵值了？?
因為在Shuffle過程將面臨一個問題，所有具有相同單詞的記錄(由單詞、URL和詞頻組成)應該交由同一個Reducer處理?
所以重新把單詞設置為鍵可以使用MapReduce框架默認的Shuffle過程,將相同單詞的所有記錄發送給同一個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:1

2.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;

最后所有的輸出結果如下

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;

下面是整個源代碼

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是用來把字符串截取成一個個標記或單詞的，默認是空格或多個空格(\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 {//統計詞頻int sum = 0;for (Text value : values) {sum += Integer.parseInt(value.toString() );}int splitIndex = key.toString().indexOf(":");//重新設置value值由URI和詞頻組成info.set( key.toString().substring( splitIndex + 1) +":"+sum );//重新設置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);//實現map函數，根據輸入的<key,value>對生成中間結果。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);//我把那兩個文件上傳到這個index目錄下了FileInputFormat.addInputPath(job, new Path("hdfs://192.168.52.140:9000/index/"));//把結果輸出到out_index+時間戳的目錄下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();}} }

轉載自：? ?https://blog.csdn.net/tanggao1314/article/details/51340672

總結

以上是生活随笔為你收集整理的Hadoop 之 MapReduce 的工作原理及其倒排索引的建立的全部內容，希望文章能夠幫你解決所遇到的問題。

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