首先要說明的是，關於老師給的實驗要求，我在網上看到了原文，原文地址：https://blog.csdn.net/qq_41035588/article/details/90514824，有興趣的同學可以去看一下。本篇博客是在此基礎上進行的重寫。
剛剛按照老師給的實驗要求完成了一次關於MapReduce的實驗，說是完成，其實也就是按照老師給的程序教程去配置關於MapReduce的一些環境，學習一些關於MapReduce的基本操作，並且學會搭建和運行MapReduce程序。現在將實驗內容和代碼按照要求展示一下，並按照我的實現步驟進行更改。
實驗目的
1.准確理解Mapreduce的設計原理
2.熟練掌握WordCount程序代碼編寫
3.學會自己編寫WordCount程序進行詞頻統計
實驗原理
MapReduce采用的是“分而治之”的思想，把對大規模數據集的操作，分發給一個主節點管理下的各個從節點共同完成，然后通過整合各個節點的中間結果，得到最終結果。簡單來說，MapReduce就是”任務的分解與結果的匯總“。
1.MapReduce的工作原理
在分布式計算中，MapReduce框架負責處理了並行編程里分布式存儲、工作調度，負載均衡、容錯處理以及網絡通信等復雜問題，現在我們把處理過程高度抽象為Map與Reduce兩個部分來進行闡述，其中Map部分負責把任務分解成多個子任務，Reduce部分負責把分解后多個子任務的處理結果匯總起來，具體設計思路如下。
（1）Map過程需要繼承org.apache.hadoop.mapreduce包中Mapper類，並重寫其map方法。通過在map方法中添加兩句把key值和value值輸出到控制台的代碼，可以發現map方法中輸入的value值存儲的是文本文件中的一行（以回車符為行結束標記），而輸入的key值存儲的是該行的首字母相對於文本文件的首地址的偏移量。然后用StringTokenizer類將每一行拆分成為一個個的字段，把截取出需要的字段（本實驗為買家id字段）設置為key，並將其作為map方法的結果輸出。
（2）Reduce過程需要繼承org.apache.hadoop.mapreduce包中Reducer類，並重寫其reduce方法。Map過程輸出的<key,value>鍵值對先經過shuffle過程把key值相同的所有value值聚集起來形成values，此時values是對應key字段的計數值所組成的列表，然后將<key,values>輸入到reduce方法中，reduce方法只要遍歷values並求和，即可得到某個單詞的總次數。
在main()主函數中新建一個Job對象，由Job對象負責管理和運行MapReduce的一個計算任務，並通過Job的一些方法對任務的參數進行相關的設置。本實驗是設置使用將繼承Mapper的doMapper類完成Map過程中的處理和使用doReducer類完成Reduce過程中的處理。還設置了Map過程和Reduce過程的輸出類型：key的類型為Text，value的類型為IntWritable。任務的輸出和輸入路徑則由字符串指定，並由FileInputFormat和FileOutputFormat分別設定。完成相應任務的參數設定后，即可調用job.waitForCompletion()方法執行任務，其余的工作都交由MapReduce框架處理。
2.MapReduce框架的作業運行流程
（1）ResourceManager：是YARN資源控制框架的中心模塊，負責集群中所有資源的統一管理和分配。它接收來自NM(NodeManager)的匯報，建立AM，並將資源派送給AM(ApplicationMaster)。
（2）NodeManager：簡稱NM，NodeManager是ResourceManager在每台機器上的代理，負責容器管理，並監控他們的資源使用情況（cpu、內存、磁盤及網絡等），以及向ResourceManager提供這些資源使用報告。
（3）ApplicationMaster：以下簡稱AM。YARN中每個應用都會啟動一個AM，負責向RM申請資源，請求NM啟動Container，並告訴Container做什么事情。
（4）Container：資源容器。YARN中所有的應用都是在Container之上運行的。AM也是在Container上運行的，不過AM的Container是RM申請的。Container是YARN中資源的抽象，它封裝了某個節點上一定量的資源（CPU和內存兩類資源）。Container由ApplicationMaster向ResourceManager申請的，由ResouceManager中的資源調度器異步分配給ApplicationMaster。Container的運行是由ApplicationMaster向資源所在的NodeManager發起的，Container運行時需提供內部執行的任務命令（可以是任何命令，比如java、Python、C++進程啟動命令均可）以及該命令執行所需的環境變量和外部資源（比如詞典文件、可執行文件、jar包等）。
另外，一個應用程序所需的Container分為兩大類，如下：
①運行ApplicationMaster的Container：這是由ResourceManager（向內部的資源調度器）申請和啟動的，用戶提交應用程序時，可指定唯一的ApplicationMaster所需的資源。
②運行各類任務的Container：這是由ApplicationMaster向ResourceManager申請的，並為了ApplicationMaster與NodeManager通信以啟動的。
以上兩類Container可能在任意節點上，它們的位置通常而言是隨機的，即ApplicationMaster可能與它管理的任務運行在一個節點上。
實驗內容
現有某電商網站用戶對商品的收藏數據，記錄了用戶收藏的商品id以及收藏日期，名為buyer_favorite1。
buyer_favorite1包含：買家id，商品id，收藏日期這三個字段，數據以“ ”分割，樣本數據及格式如下：
買家id 商品id 收藏日期
10181 1000481 2010-04-04 16:54:31 20001 1001597 2010-04-07 15:07:52 20001 1001560 2010-04-07 15:08:27 20042 1001368 2010-04-08 08:20:30 20067 1002061 2010-04-08 16:45:33 20056 1003289 2010-04-12 10:50:55 20056 1003290 2010-04-12 11:57:35 20056 1003292 2010-04-12 12:05:29 20054 1002420 2010-04-14 15:24:12 20055 1001679 2010-04-14 19:46:04 20054 1010675 2010-04-14 15:23:53 20054 1002429 2010-04-14 17:52:45 20076 1002427 2010-04-14 19:35:39 20054 1003326 2010-04-20 12:54:44 20056 1002420 2010-04-15 11:24:49 20064 1002422 2010-04-15 11:35:54 20056 1003066 2010-04-15 11:43:01 20056 1003055 2010-04-15 11:43:06 20056 1010183 2010-04-15 11:45:24 20056 1002422 2010-04-15 11:45:49 20056 1003100 2010-04-15 11:45:54 20056 1003094 2010-04-15 11:45:57 20056 1003064 2010-04-15 11:46:04 20056 1010178 2010-04-15 16:15:20 20076 1003101 2010-04-15 16:37:27 20076 1003103 2010-04-15 16:37:05 20076 1003100 2010-04-15 16:37:18 20076 1003066 2010-04-15 16:37:31 20054 1003103 2010-04-15 16:40:14 20054 1003100 2010-04-15 16:40:1
要求編寫MapReduce程序，統計每個買家收藏商品數量。
實驗步驟
MapReduce環境配置：
1、下載hadoop-eclipse-plugin-2.6.0 的jar包，復制到eclipse中的plugins目錄下（對於較高版本的eclipse擁有dropins目錄，需要將jar包放在這里，一定不要放在plugins目錄下）。
2、啟動eclipse，點擊Window -> Show View -> Other,選擇Map/Reduce Locations ,然后在下面控制台的位置出現Map/Reduce Locations.
3、右擊空白處，點擊New Hadoop location,按照下圖進行配置：
其中，Location Name可以隨便起一個名字；Map/Reduce Master為你hdfs.site-xml中的端口號，Host為你的端口號或者是你的host名字；DFS Master為你core.site-xml中的端口號（一般的配置應該都是和我一樣的）。
4、配置環境變量。在系統變量中添加HADOOP_USER_NAME，內容為在配置過程中的location name，如下如所示:
實現流程與代碼：
1、啟動在linux上hadoop
$HADOOP_HOME/sbin/start-all.sh
2.在Windows上，創建一個文本文件buyer_favorite1，並將上面給出的數據粘貼到buyer_favorite1文件中（要注意文本格式，因為MapReduce是按行讀取的，所以最后一行一定不要有空格）。
3、在DFS Location中創建目錄
右擊，將文件上傳至testhdfs1026/run/input中，此時在50070界面上是可以查看到的。
4、新建MapReduce項目，名字為mapreduce1,創建包Mapreduce，類WordCount
5、將以下代碼復制進去
package mapreduce; import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; 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; public class WordCount { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Job job = Job.getInstance(); job.setJobName("WordCount"); job.setJarByClass(WordCount.class); job.setMapperClass(doMapper.class); job.setReducerClass(doReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); Path in = new Path("hdfs://192.168.57.128:9000/testhdfs1026/run/input/buyer_favorite1.txt"); Path out = new Path("hdfs://192.168.57.128:9000/testhdfs1026/run/output/buyer_favorite1"); FileInputFormat.addInputPath(job, in); FileOutputFormat.setOutputPath(job, out); boolean flag=job.waitForCompletion(true); System.out.println(flag); System.exit( flag? 0 : 1); } public static class doMapper extends Mapper<Object, Text, Text, IntWritable> { public static final IntWritable one = new IntWritable(1); public static Text word = new Text(); @Override protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer tokenizer = new StringTokenizer(value.toString(), " "); word.set(tokenizer.nextToken()); context.write(word, one); } } public static class doReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable value : values) { sum += value.get(); } result.set(sum); context.write(key, result); } } }
6、在WordCount類文件中，單擊右鍵=>Run As=>Run on Hadoop選項，將MapReduce任務提交到Hadoop中，等待運行。
實驗截圖：
代碼基本構思：
大致思路是將hdfs上的文本作為輸入，MapReduce通過InputFormat會將文本進行切片處理，並將每行的首字母相對於文本文件的首地址的偏移量作為輸入鍵值對的key，文本內容作為輸入鍵值對的value，經過在map函數處理，輸出中間結果<word,1>的形式，並在reduce函數中完成對每個單詞的詞頻統計。整個程序代碼主要包括兩部分：Mapper部分和Reducer部分。
Mapper代碼
public static class doMapper extends Mapper<Object, Text, Text, IntWritable>{ //第一個Object表示輸入key的類型；第二個Text表示輸入value的類型；第三個Text表示表示輸出鍵的類型；第四個IntWritable表示輸出值的類型 public static final IntWritable one = new IntWritable(1); public static Text word = new Text(); @Override protected void map(Object key, Text value, Context context) throws IOException, InterruptedException //拋出異常 { StringTokenizer tokenizer = new StringTokenizer(value.toString(),"\t"); //StringTokenizer是Java工具包中的一個類，用於將字符串進行拆分 word.set(tokenizer.nextToken()); //返回當前位置到下一個分隔符之間的字符串 context.write(word, one); //將word存到容器中，記一個數 }
在map函數里有三個參數，前面兩個Object key,Text value就是輸入的key和value，第三個參數Context context是可以記錄輸入的key和value。例如context.write(word,one)；此外context還會記錄map運算的狀態。map階段采用Hadoop的默認的作業輸入方式，把輸入的value用StringTokenizer()方法截取出的買家id字段設置為key，設置value為1，然后直接輸出<key,value>。
Reducer代碼
public static class doReducer extends Reducer<Text, IntWritable, Text, IntWritable>{ //參數同Map一樣，依次表示是輸入鍵類型，輸入值類型，輸出鍵類型，輸出值類型 private IntWritable result = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable value : values) { sum += value.get(); } //for循環遍歷，將得到的values值累加 result.set(sum); context.write(key, result); } }
map輸出的<key,value>先要經過shuffle過程把相同key值的所有value聚集起來形成<key,values>后交給reduce端。reduce端接收到<key,values>之后，將輸入的key直接復制給輸出的key,用for循環遍歷values並求和，求和結果就是key值代表的單詞出現的總次，將其設置為value，直接輸出<key,value>。
本次測試到此為止，下次帶來海量數據的存儲統計。