分布式并行计算MapReduce

落日映苍穹つ 2021-12-10 13:53 321阅读 0赞

#### 1、用自己的话阐明Hadoop平台上HDFS和MapReduce的功能、工作原理和工作过程。 ####

MapReduce是我们再进行离线大数据处理的时候经常要使用的计算模型，MapReduce的计算过程被封装的很好，我们只用使用Map和Reduce函数，所以对其整体的计算过程不是太清楚，同时MapReduce1.0和MapReduce2.0在网上有很多人混淆。

## MapReduce1.0运行模型 ##

![Image 1][]

20170730014216035.png

#### Input ####

Input但是输入文件的存储位置，

###### 但是注意这里并一定是一些博客说的当然是HDFS似的分布式文件系统位置，默认是HDFS文件系统，当然也可以修改。 ######

，它也可以是本机上的文件位置。  
我们来仔细分析下input

![Image 1][]

8aab5880-d171-30f7-91d6-aaacba2d03ce.jpg

首先我们知道要和JobTracker打交道是离不开JobClient这个接口的，就如上图所示，

然后JobClient中的Run方法 会让 JobClient 把所有 Hadoop Job 的信息，比如 mapper reducer jar path, mapper / reducer class name, 输入文件的路径等等，告诉给 JobTracker，如下面的代码所示：

public int run(String[] args) throws Exception { //create job Job job = Job.getInstance(getConf(), this.getClass().getSimpleName()); // set run jar class job.setJarByClass(this.getClass()); // set input . output FileInputFormat.addInputPath(job, new Path(PropReader.Reader("arg1"))); FileOutputFormat.setOutputPath(job, new Path(PropReader.Reader("arg2"))); // set map job.setMapperClass(HFile2TabMapper.class); job.setMapOutputKeyClass(ImmutableBytesWritable.class); job.setMapOutputValueClass(Put.class); // set reduce job.setReducerClass(PutSortReducer.class); return 0; }

除此以外，JobClient.runJob() 还会做一件事，使用 InputFormat类去计算如何把 input 文件 分割成一份一份，然后交给 mapper 处理。inputformat.getSplit() 函数返回一个 InputSplit 的 List, 每一个 InputSplit 就是一个 mapper 需要处理的数据。

一个 Hadoop Job的 input 既可以是一个很大的 file, 也可以是多个 file; 无论怎样，getSplit() 都会计算如何分割 input.

###### 如果是HDFS文件系统，我们都知道其可以通过将文件分割为block的形式存放在很多台电脑上，使其可以存放很大的文件。那么Mapper是如何确定一个HDFS文件中的block存放哪几台电脑，有什么数据？ ######

inputFormat它实际上是个 interface, 需要 类 来继承，提供分割 input 的逻辑。

Jobclient 有一个方法叫 setInputFormat(), 通过它，我们可以告诉 JobTracker 想要使用的 InputFormat 类 是什么。如果我们不设置，Hadoop默认的是 TextInputFormat, 它默认为文件在 HDFS上的每一个 Block 生成一个对应的 InputSplit. 所以大家使用 Hadoop 时，也可以编写自己的 input format, 这样可以自由的选择分割 input 的算法，甚至处理存储在 HDFS 之外的数据。

##### JobTracker 尽量把 mapper 安排在离它要处理的数据比较近的机器上，以便 mapper 从本机读取数据，节省网络传输时间。具体实现是如何实现？ #####

对于每个 map任务, 我们知道它的 split 包含的数据所在的主机位置，我们就把 mapper 安排在那个相应的主机上好了，至少是比较近的host. 你可能会问：split 里存储的 主机位置是 HDFS 存数据的主机，和 MapReduce 的主机 有什么相关呢？为了达到数据本地性，其实通常把MapReduce 和 HDFS 部署在同一组主机上。

既然一个 InputSplit 对应一个 map任务, 那么当 map 任务收到它所处理数据的位置信息，它就可以从 HDFS 读取这些数据了。

###### 接下来我们再从map函数看Input ######

map函数接受的是一个 key value 对。

###### 实际上，Hadoop 会把每个 mapper 的输入数据再次分割，分割成一个个 key-value对, 然后为每一个 key-value对，调用Map函数一次. 为了这一步分割，Hadoop 使用到另一个类: RecordReader. 它主要的方法是 next(), 作用就是从 InputSplit 读出一条 key-value对. ######

RecordReader 可以被定义在每个 InputFormat 类中。当我们通过 JobClient.setInputFormat() 告诉 Hadoop inputFormat 类名称的时候， RecordReader 的定义也一并被传递过来。

所以整个Input,

1.JobClient输入输入文件的存储位置

2.JobClient通过InputFormat接口可以设置分割的逻辑，默认是按HDFS文件分割。

3.Hadoop把文件再次分割为key-value对。

4.JobTracker负责分配对应的分割块由对应的maper处理，同时 RecordReader负责读取key-value对值。

#### Mapper ####

JobClient运行后获得所需的配置文件和客户端计算所得的输入划分信息。并将这些信息都存放在JobTracker专门为该作业创建的文件夹中。文件夹名为该作业的Job ID。JAR文件默认会有10个副本（mapred.submit.replication属性控制）；

然后输入划分信息告诉了JobTracker应该为这个作业启动多少个map任务等信息。

JobTracker通过TaskTracker 向其汇报的心跳情况和slot（情况），每一个slot可以接受一个map任务，这样为了每一台机器map任务的平均分配，JobTracker会接受每一个TaskTracker所监控的slot情况。

JobTracker接收到作业后，将其放在一个作业队列里，等待作业调度器对其进行调度，当作业调度器根据自己的调度算法调度到该作业时，会根据输入划分信息为每个划分创建一个map任务，并将map任务分配给TaskTracker执行，分配时根据slot的情况作为标准。

TaskTracker每隔一段时间会给JobTracker发送一个心跳，告诉JobTracker它依然在运行，同时心跳中还携带着很多的信息，比如当前map任务完成的进度等信息。当JobTracker收到作业的最后一个任务完成信息时，便把该作业设置成“成功”。当JobClient查询状态时，它将得知任务已完成，便显示一条消息给用户。

###### Map通过 RecordReader 读取Input的key/value对，map根据用户自定义的任务，运行完毕后，产生另外一系列 key/value，并将其写入到Hadoop的内存缓冲取中，在内存缓冲区中的key/value对按key排序，此时会按照reduce partition进行，分到不同partition中，一旦内存满就会被写入到本地磁盘的文件里，这个文件叫spill file。 ######

#### shuffle ####

Shuffle是我们不需要编写的模块，但却是十分关键的模块。

![Image 1][]

4df193f5-e56e-308f-9689-eac035dd8a2b.png

在map中，每个 map 函数会输出一组 key/value对, Shuffle 阶段需要从所有 map主机上把相同的 key 的 key value对组合在一起，（也就是这里省去的Combiner阶段）组合后传给 reduce主机, 作为输入进入 reduce函数里。

Partitioner组件 负责计算哪些 key 应当被放到同一个 reduce 里

HashPartitioner类，它会把 key 放进一个 hash函数里，然后得到结果。如果两个 key 的哈希值 一样，他们的 key/value对 就被放到同一个 reduce 函数里。我们也把分配到同一个 reduce函数里的 key /value对 叫做一个reduce partition.

我们看到 hash 函数最终产生多少不同的结果, 这个 Hadoop job 就会有多少个 reduce partition／reduce 函数，这些 reduce函数最终被JobTracker 分配到负责 reduce 的主机上，进行处理。

我们知道map阶段可能会产生多个spill file 当 Map 结束时，这些 spill file 会被 merge 起来,不是 merge 成一个 file，而是也会按 reduce partition 分成多个。

当 Map tasks 成功结束时，他们会通知负责的 tasktracker, 然后消息通过 jobtracker 的 heartbeat 传给 jobtracker. 这样，对于每一个 job, jobtracker 知道 map output 和 map tasks 的关联。Reducer 内部有一个 thread 负责定期向 jobtracker 询问 map output 的位置，直到 reducer 得到所有它需要处理的 map output 的位置。

Reducer 的另一个 thread 会把拷贝过来的 map output file merge 成更大的 file. 如果 map task 被 configure 成需要对 map output 进行压缩，那 reduce 还要对 map 结果进行解压缩。当一个 reduce task 所有的 map output 都被拷贝到一个它的 host上时，reduce 就要开始对他们排序了。

排序并不是一次把所有 file 都排序，而是分几轮。每轮过后产生一个结果，然后再对结果排序。最后一轮就不用产生排序结果了，而是直接向 reduce 提供输入。这时，用户提供的 reduce函数 就可以被调用了。输入就是 map 任务 产生的 key value对.

###### 同时reduce任务并不是在map任务完全结束后才开始的，Map 任务有可能在不同时间结束，所以 reduce 任务没必要等所有 map任务 都结束才开始。事实上，每个 reduce任务有一些 threads 专门负责从 map主机复制 map 输出（默认是5个）。 ######

#### Reduce ####

![Image 1][]

reduce() 函数以 key 及对应的 value 列表作为输入，按照用户自己的程序逻辑，经合并 key 相同的 value 值后，产 生另外一系列 key/value 对作为最终输出写入 HDFS

#### 2.HDFS上运行MapReduce ####

#### 1）查看是否已经安装python： ####

![1480639-20190621090954076-436690210.png][]

#### 2）在/home/hadoop/路径下建立wc文件夹，在文件夹内新建mapper.py、reducer.py、run.sh和文本文件HarryPotter.txt： ####

![1480639-20190621091249662-1713364768.png][]

#### 3）查看mapper.py reducer.py run.sh的内容： ####

![1480639-20190621091725011-107468679.png][]

![1480639-20190621091725279-2142473329.png][]

#### 4）修改mapper.py和reducer.py文件的权限： ####

![1480639-20190621091838155-2072905864.png][]

#### 5）测试mapper.py和reducer.py： ####

![1480639-20190621092109117-1218393895.png][]

#### 6）启动hadoop： ####

![1480639-20190621092328419-1775546482.png][]

#### 7）把文本文件上传到hdfs： ####

![1480639-20190621092445535-1567994355.png][]

#### 8）将hadoop-streaming-2.7.1.jar的路径添加到bashrc文件并且让环境变量生效： ####

![1480639-20190621092504642-487761250.png][]

![1480639-20190621092648731-1942081230.png][]

#### 9）运行run.sh文件统计文本： ####

![1480639-20190621093050224-1509342817.png][]

![1480639-20190621092904206-471393097.png][]

转载于:https://www.cnblogs.com/shang1680/p/10966620.html

[Image 1]: 
[1480639-20190621090954076-436690210.png]: /images/20211210/a09e436ed3ef447482159f3857389f43.png
[1480639-20190621091249662-1713364768.png]: /images/20211210/2b3de1cb7aec4cdd84f0aca349371527.png
[1480639-20190621091725011-107468679.png]: /images/20211210/9187dc1a3ab74ae69572242758fa1570.png
[1480639-20190621091725279-2142473329.png]: /images/20211210/9c9e49d3370e438e9f7d490c831b91bf.png
[1480639-20190621091838155-2072905864.png]: /images/20211210/56450da5d94043c6b38b8ff3aff0951f.png
[1480639-20190621092109117-1218393895.png]: /images/20211210/d6179f75f13045908f7fce19363397e9.png
[1480639-20190621092328419-1775546482.png]: /images/20211210/df8d04e1b80a4b84ae5acfcf363dc1d6.png
[1480639-20190621092445535-1567994355.png]: /images/20211210/b17fef32822e4f61b0ce342ffd165761.png
[1480639-20190621092504642-487761250.png]: /images/20211210/3d12adb3ab0a4ad9a0f07be33cdbf015.png
[1480639-20190621092648731-1942081230.png]: /images/20211210/181603f27b694152acacd3977e69baf3.png
[1480639-20190621093050224-1509342817.png]: /images/20211210/ff66a0938f68450f8d3484c10aaae3e5.png
[1480639-20190621092904206-471393097.png]: /images/20211210/36f11058ea6a4580b05c3362b5ab8bb9.png