分布式并行计算MapReduce

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
16
1import org.apache.hadoop.io.IntWritable;

2import org.apache.hadoop.io.LongWritable;

3import org.apache.hadoop.io.Text;

4import org.apache.hadoop.mapreduce.Mapper;

5

6import java.io.IOException;

7

8public class Map extends Mapper<LongWritable, Text, Text,IntWritable> {

9    private final static IntWritable one = new IntWritable(1);

10    private Text word = new Text();

11    public void map(LongWritable key,Text value,Context context) throws IOException, InterruptedException {

12        word.set(value.toString());

13        context.write(word, one);

14    }

15}

16

reduce.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
1import org.apache.hadoop.io.IntWritable;

2import org.apache.hadoop.io.Text;

3import org.apache.hadoop.mapreduce.Reducer;

4

5import java.io.IOException;

6import java.util.Iterator;

7

8public class Reduce extends Reducer<Text, IntWritable, Text,IntWritable> {

9    @Override

10    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {

11        int sum = 0;

12        for(IntWritable intWritable : values){

13            sum += intWritable.get();

14        }

15        context.write(key, new IntWritable(sum));

16    }

17}

18

1
2
3
4
5
6
1#!/usr/bin/env python

2cd /home/hadoop/wc

3sudo gedit reduce.py

4#赋予权限

5chmod a+x /home/hadoop/map.py

6

本机上测试运行代码：

1
2
3
4
1echo "foo foo quux labs foo bar quux" | /home/hadoop/wc/mapper.py

2

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

4

1
2
3
1启动Hadoop，HDFS, JobTracker, TaskTracker：  

2

3

放到HDFS上运行

下载并上传文件到hdfs上：

1
2
3
4
5
6
7
8
9
1#上传文件

2cd  /home/hadoop/wc

3wget http://www.gutenberg.org/files/5000/5000-8.txt

4wget http://www.gutenberg.org/cache/epub/20417/pg20417.txt

5 

6#下载文件

7cd /usr/hadoop/wc

8hdfs dfs -put /home/hadoop/hadoop/gutenberg/*.txt /user/hadoop/input

9

新建一个文件5000-8.txt，运行结果如下：