【大数据应用技术】做业十一｜分布式并行计算MapReduce

本次做业在要求来自：https://edu.cnblogs.com/campus/gzcc/GZCC-16SE2/homework/3319

 1.用本身的话阐明Hadoop平台上HDFS和MapReduce的功能、工做原理和工做过程。

1）HDFS

HDFS是分布式文件系统，用来存储海量数据。HDFS中有两类节点：NameNode和DataNode。

NameNode是管理节点，存放文件元数据。也就是存放着文件和数据块的映射表，数据块和数据节点的映射表。也就是说，经过NameNode，咱们就能够找到文件存放的地方，找到存放的数据。DataNode是工做节点，用来存放数据块，也就是文件实际存储的地方。

工做原理：客户端向NameNode发起读取元数据的消息，NameNode就会查询它的Block Map，找到对应的数据节点。而后客户端就能够去对应的数据节点中找到数据块，拼接成文件就能够了，这就是读写的流程。

2）MapReduce

MapReduce是并行处理框架，实现任务分解和调度。

工做原理：将一个大任务分解成多个小任务(map)，小任务执行完了以后，合并计算结果(reduce)。也就是说，JobTracker拿到job以后，会把job分红不少个maptask和reducetask，交给他们执行。 MapTask、ReduceTask函数的输入、输出都是的形式。HDFS存储的输入数据通过解析后，以键值对的形式，输入到MapReduce()函数中进行处理，输出一系列键值对做为中间结果，在Reduce阶段，对拥有一样Key值的中间数据进行合并造成最后结果。

2.HDFS上运行MapReduce

1）准备文本文件，放在本地/home/hadoop/wc

先准备一个大一点英文文本文件，我这里准备的是一个名为Mrstandfast.txt的文本文件，放在了下载目录下，以下图所示。

使用 mkdir wc 命令新建一个名为wc的文件夹，再使用 mv /home/chen/下载/MrStandfast.txt MrStandfast.txt 命令把MrStandfast.txt文件复制到wc中，以下图所示。

2）编写mapper函数和reducer函数，在本地运行测试经过

首先，咱们能够先编写 mapper函数和reducer函数，使用 gedit mapper.py 命令创建mapper.py文件，在其中插入所要执行的语句，并保存关闭。同理，reducer.py也是这样。

mapper.py

#!/usr/bin/env python
import sys for line in sys.stdin: line = line.strip() words = line.split() for word in words: print "%s\t%s" % (word, 1)

 

reducer.py

#!/usr/bin/env python
 
from operator import itemgetter import sys current_word = None current_count = 0 word = None # input comes from STDIN
for line in sys.stdin: # remove leading and trailing whitespace
    line = line.strip() # parse the input we got from mapper.py
    word, count = line.split('\t', 1) # convert count (currently a string) to int
    try: count = int(count) except ValueError: # count was not a number, so silently
        # ignore/discard this line
        continue
 
    # this IF-switch only works because Hadoop sorts map output
    # by key (here: word) before it is passed to the reducer
    if current_word == word: current_count += count else: if current_word: # write result to STDOUT
            print '%s\t%s' % (current_word, current_count) current_count = count current_word = word # do not forget to output the last word if needed!
if current_word == word: print '%s\t%s' % (current_word, current_count)

完成上述步骤后，可使用 cat mapper.py 命令和 cat reducer.py 命令来查看，以下图所示。

分别使用 chmod a+x /home/chen/wc/mapper.py 和 chmod a+x /home/chen/wc/reducer.py 命令修改mapper和reducer的权限。

分别使用 echo "foo foo quux labs foo bar quux" | ./mapper 命令和 echo "foo foo quux labs foo bar quux" | ./mapper.py | sort -k1,1 | ./reducer.py 命令在本地测试python代码是否可执行，以下图所示。

3）启动Hadoop：HDFS, JobTracker, TaskTracker

 使用 start-all.sh 命令启动hadoop，再使用 jps 命令查看是否启动成功，以下图所示。

4）把文本文件上传到hdfs文件系统上 user/chen/input 

 因为我先前作实验时已经建立了 /user/chen/input 这个文件夹了，因此在这里我就直接将本地文件上传便可，使用 hdfs dfs -put /home/chen/wc/MrStandfast.txt /user/chen/input/ 命令把本地的MrStandfast.txt上传至hdfs文件系统上 user/chen/input上，再使用 hdfs dfs -ls /user/chen/input/ 命令来查看文件，以下图所示。

注意：若是先前尚未建立文件夹的，可使用 hdfs dfs -mkdir -p /user/chen/input 命令来建立文件夹，详见https://www.cnblogs.com/bhuan/p/10964927.html

5）streaming的jar文件的路径写入环境变量，让环境变量生效

使用 vim ~/.bashrc 命令将streaming的jar文件的路径写入~/.bashrc中，并使用 source ~/.bashrc 让环境变量生效，以下图所示。

streaming的jar文件的路径：

export STREAM=$HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar

6）创建一个shell脚本文件：streaming接口运行的脚本，名称为run.sh

使用 vim run.sh 命令或者 gedit run.sh 命令添加streaming接口运行的脚本，再使用 source run.sh 命令使其生效，以下图所示。

run.sh文件内容

hadoop jar $STREAM  \
-file /home/chen/wc/mapper.py \
-mapper  /home/chen/wc/mapper.py \
-file /home/chen/wc/reducer.py \
-reducer  /home/chen/wc/reducer.py \
-input /user/chen/input/*.txt \
-output /user/chen/wcountput

7）source run.sh来执行mapreduce

8）查看运行结果

使用 hdfs dfs -cat /user/chen/wcountput/* 命令来查看运行结果，以下图所示。