分布式並行計算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

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

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));
    }
}

#!/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上：

#上傳文件
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，運行結果如下：