分佈式並行計算MapReduce

這個做業的要求來自於：https://edu.cnblogs.com/campus/gzcc/GZCC-16SE2/homework/3319。

1.用本身的話闡明Hadoop平臺上HDFS和MapReduce的功能、工做原理和工做過程。

HDFS

    功能：分佈式文件系統，用來存儲海量數據。

    工做原理和過程：HDFS是Hadoop的分佈式文件系統，HDFS中的文件會默認存儲3份，存儲在不一樣的機器上，提供容錯機制，副本丟失或者宕機的自動恢復。HDFS整體上採用Master/Slave的架構，整個HDFS架構由Client、NameNode、Secondary NameNode和DataNode構成。NameNode負責存儲整個集羣的元數據信息，Client能夠根據元數據信息找到對應的文件，DataNode負責數據的實際存儲。當一個文件上傳到HDFS的時候，DataNode會按照Block爲基本單位分佈在各個DataNode中，並且爲了保護數據的一致性和容錯性，通常一份數據會在不一樣的DataNode上默認存儲三份。以下圖所示：

 

MapReduce

    功能：並行處理框架，實現任務分解和調度。

    工做原理和過程：MapReduce的工做過程分紅兩個階段，map階段和reduce階段。每一個階段都有鍵值對做爲輸入輸出，map函數和reduce函數的具體實現由程序員完成。MapReduce的框架也是採用Master/Slave的方式組織，以下圖所示。由四部分組成，分別爲Client、JobTracker、TaskTracker以及Task。JobTracker主要負責資源監控和做業調度。JobTracker監控TaskTracker是否存活，任務執行的狀態以及資源的使用狀況，而且把獲得的信息交給TaskSceduler。TaskSceduler根據每一個TaskTracker的狀況給分配響應的任務。TaskTracker會週期性經過heartbeats向JobTracker發送資源的使用狀況，任務的執行情況等信息，同時會接收JobTracker的指令，TaskTracker把本身可支配的資源分紅若干個Slot，Task只有拿到一個Slot資源才能執行任務。Task任務分紅Map Task和Reduce Task兩種任務，都是由TaskTracker進行調度的。

 

 

2.HDFS上運行MapReduce

mapper.py

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

import  org.apache.hadoop.io.IntWritable; import  org.apache.hadoop.io.LongWritable; import  org.apache.hadoop.io.Text; import  org.apache.hadoop.mapreduce.Mapper;   import  java.io.IOException;   public  class  Map  extends Mapper<LongWritable, Text, Text,IntWritable> {      private final static IntWritable one  =  new IntWritable( 1 );      private Text word  =  new Text();      public void  map (LongWritable key,Text value,Context context) throws IOException, InterruptedException {          word. set (value.toString());          context.write(word, one);      } }

reduce.py

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

import  org.apache.hadoop.io.IntWritable; import  org.apache.hadoop.io.Text; import  org.apache.hadoop.mapreduce.Reducer;   import  java.io.IOException; import  java.util.Iterator;   public  class  Reduce  extends Reducer<Text, IntWritable, Text,IntWritable> {      @Override      protected void  reduce (Text key, Iterable values, Context context) throws IOException, InterruptedException {          int  sum  =  0 ;          for (IntWritable intWritable : values){              sum  + =  intWritable.get();          }          context.write(key, new IntWritable( sum ));      } }

1 2 3 4 5

#!/usr/bin/env python cd  / home / hadoop / wc sudo gedit  reduce .py #賦予權限 chmod a + x  / home / hadoop / map .py

本機上測試運行代碼：

echo "foo foo quux labs foo bar quux" | /home/hadoop/wc/mapper.py

echo "foo foo quux labs foo bar quux" | /home/hadoop/wc/mapper.py | sort -k1,1 | /home/hadoop/wc/reducer.p

啓動Hadoop，HDFS, JobTracker, TaskTracker：

放到HDFS上運行

下載並上傳文件到hdfs上：

1 2 3 4 5 6 7 8

#上傳文件 cd   / home / hadoop / wc wget http: / / www.gutenberg.org / files / 5000 / 5000 - 8.txt wget http: / / www.gutenberg.org / cache / epub / 20417 / pg20417.txt    #下載文件 cd  / usr / hadoop / wc hdfs dfs  - put  / home / hadoop / hadoop / gutenberg / * .txt  / user / hadoop / input

新建一個文件5000-8.txt，運行結果以下：