MapReduce介绍以及原理_mapreduce的思想原理

什么是MapReduce

MapReduce是一种编程模型，用于大规模数据集（大于1TB）的并行运算。概念"Map（映射）“和"Reduce（归约）”，是它们的主要思想，都是从函数式编程语言里借来的，还有从矢量编程语言里借来的特性。它极大地方便了编程人员在不会分布式并行编程的情况下，将自己的程序运行在分布式系统上。 当前的软件实现是指定一个Map（映射）函数，用来把一组键值对映射成一组新的键值对，指定并发的Reduce（归约）函数，用来保证所有映射的键值对中的每一个共享相同的键组
MapReduce是一个分布式计算框架:
它将大型数据操作作业分解为可以跨服务器集群并行执行的单个任务。起源于Google。
适用于大规模数据处理场景
每个节点处理存储在该节点的数据
每个job包含Map和Reduce两部分
MapReduce是面向大数据并行处理的计算模型、框架和平台，它隐含了以下三层含义：
1）MapReduce是一个基于集群的高性能并行计算平台（Cluster Infrastructure）。它允许用市场上普通的商用服务器构成一个包含数十、数百至数千个节点的分布和并行计算集群。
2）MapReduce是一个并行计算与运行软件框架（Software Framework）。它提供了一个庞大但设计精良的并行计算软件框架，能自动完成计算任务的并行化处理，自动划分计算数据和计算任务，在集群节点上自动分配和执行任务以及收集计算结果，将数据分布存储、数据通信、容错处理等并行计算涉及到的很多系统底层的复杂细节交由系统负责处理，大大减少了软件开发人员的负担。
3）MapReduce是一个并行程序设计模型与方法（Programming Model & Methodology）。它借助于函数式程序设计语言Lisp的设计思想，提供了一种简便的并行程序设计方法，用Map和Reduce两个函数编程实现基本的并行计算任务，提供了抽象的操作和并行编程接口，以简单方便地完成大规模数据的编程和计算处理 。

MapReduce的设计思想

大规模数据处理时，MapReduce在三个层面上的基本构思

1如何对付大数据处理：分而治之

一个大数据若可以分为具有同样计算过程的数据块，并且这些数据块之间不存在数据依赖关系，则提高处理速度的最好办法就是采用“分而治之”的策略进行 并行化计算。MapReduce采用了这种“分而治之”的设计思想，对相互间不具有或者有较少数据依赖关系的大数据，用一定的数据划分方法对数据分片，然 后将每个数据分片交由一个节点去处理，最后汇总处理结果。

2上升到抽象模型：Mapper与Reducer

MPI等并行计算方法缺少高层并行编程模型，为了克服这一缺陷，MapReduce借鉴了Lisp函数式语言中的思想，用Map和Reduce两个函数提供了高层的并行编程抽象模型和接口，程序员只要实现这两个基本接口即可快速完成并行化程序的设计。
与Lisp语言可以用来处理列表数据一样，MapReduce的设计目标是可以对一组顺序组织的数据元素/记录进行处理。现实生活中，大数据往往是 由一组重复的数据元素/记录组成，例如，一个Web访问日志文件数据会由大量的重复性的访问日志构成，对这种顺序式数据元素/记录的处理通常也是顺序式扫 描处理。
MapReduce将以上的处理过程抽象为两个基本操作，把上述处理过程中的前两步抽象为Map操作，把后两步抽象为Reduce操作。于是Map 操作主要负责对一组数据记录进行某种重复处理，而Reduce操作主要负责对Map的中间结果进行某种进一步的结果整理和输出。以这种方 式，MapReduce为大数据处理过程中的主要处理操作提供了一种抽象机制。

3上升到构架：统一构架，为程序员隐藏系统层细节

MPI等并行计算方法缺少统一的计算框架支持，程序员需要考虑数据存储、划分、分发、结果收集、错误恢复等诸多细节；为此，MapReduce设计并提供了统一的计算框架，为程序员隐藏了绝大多数系统层面的处理细节。
1）计算任务的自动划分和调度。
2）数据的自动化分布存储和划分。
3）处理数据与计算任务的同步。
4）结果数据的收集整理（sorting，combining，partitioning，等）。
5）系统通信、负载平衡、计算性能优化处理。
6）处理系统节点出错检测和失效恢复。

MapReduce特点

优点
1、易于编程
2、可扩展性
3、高容错性
4、高吞吐量
不适用领域
1、难以实时计算
2、不适合流式计算

MapReduce实现WordCount

过程简述

上图的流程大概分为以下几步：
　　1、假设一个文件有几行英文单词作为 MapReduce 的Input（输入），这里经过 Splitting过程把文件分割为3块，分割的方式是按照各句子的hash值作为依据。分割后的3块数据就可以并行处理，每一块交给一个 map 线程处理。
　　2、每个map线程中，以每个单词为key，以1作为词频数value，然后输出。
　　3、每个map的输出要经过shuffling（混洗），将相同的单词key放在一个桶里面，然后交给reduce处理。
　　4、reduce接受到shuffle后的数据，会将相同的单词进行合并，得到每个单词的词频数，最后将统计好的每个单词的词频数作为输出结果。
　　上述就是 MapReduce 的大致流程，前两步可以看做map阶段，后两步可以看做reduce 阶段。

代码实现

pom.xml:

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>mymr</groupId>
  <artifactId>mymr</artifactId>
  <version>1.0-SNAPSHOT</version>

  <name>mymr</name>
  <!-- FIXME change it to the project's website -->
  <url>http://www.example.com</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.7</maven.compiler.source>
    <maven.compiler.target>1.7</maven.compiler.target>
  </properties>

  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>4.11</version>
      <scope>test</scope>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-hdfs -->
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-hdfs</artifactId>
      <version>2.6.0</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-common -->
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-common</artifactId>
      <version>2.6.0</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-client -->
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-client</artifactId>
      <version>2.6.0</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-core -->
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-core</artifactId>
      <version>1.2.1</version>
    </dependency>
  </dependencies>

  <build>
    <plugins>
      <plugin>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>2.3.2</version>
        <configuration>
          <source>1.8</source>
          <target>1.8</target>
        </configuration>
      </plugin>
      <plugin>
        <artifactId>maven-assembly-plugin</artifactId>
        <configuration>
          <descriptorRefs>
            <descriptorRef>jar-with-dependencies</descriptorRef>
          </descriptorRefs>
          <archive>
            <manifest>
              <mainClass>mymr.services.wc.MyDriver</mainClass>
            </manifest>
          </archive>
        </configuration>
        <executions>
          <execution>
            <id>make-assembly</id>
            <phase>package</phase>
            <goals>
              <goal>single</goal>
            </goals>
          </execution>
        </executions>
      </plugin>
    </plugins>

  </build>
</project>


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

My Driver:

package mymr.services.wc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MyDriver {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
        conf.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());
        //准备一个空任务
        Job job = Job.getInstance(conf,"wc");
        //设置该任务主启动类
        job.setJarByClass(MyDriver.class);
        //设置任务的输入数据源
        FileInputFormat.addInputPath(job,new Path("F:\\study files\\hadoop\\testfiles\\abc.txt"));
        //设置你的mapper任务类
        job.setMapperClass(MyMapper.class);
        //设置Mapper任务类的输出数据类型
        job.setOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        //设置你的Reducer任务类
        job.setReducerClass(MyReduce.class);
        //设置Reduce任务类输出的数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        //设置任务的输出视距目标
        FileOutputFormat.setOutputPath(job,new Path("F:\\study files\\hadoop\\testfiles\\test"));
        //启动任务并执行
        job.waitForCompletion(true);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

My Mapper:

package mymr.services.wc;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class MyMapper extends Mapper<LongWritable, Text,Text,LongWritable> {
    private LongWritable one = new  LongWritable(1);
    private Text word = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String [] wds = value.toString().split(" ");
        for (String word : wds) {
            Text wd = new Text(word);
            context.write(wd,one);
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

My Reduce:

package mymr.services.wc;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class MyReduce extends Reducer<Text, LongWritable,Text,LongWritable> {
    private LongWritable res = new LongWritable();

    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
        long ressum = 0;
        for (LongWritable one : values){
            ressum+=one.get();
        }
        res.set(ressum);
        context.write(key,res);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

在windows下测试时需要先在windows下配置hadoop：
1、解压hadoop压缩包

2、配置环境变量

3、在bin下添加所需winuntils支持
可在此链接获取，本人亲测这个版本在2020/6/4前可与windows版本匹配
链接：https://pan.baidu.com/s/1nSebZHlnvLiFZdoyKeDoRg 提取码：uhau

4、在C:\Windows\System32下添加hadoop.dll

配置完hadoop即可在windows下测试：
此为测试文件：

开启测试：

此为测试结果：


在linux虚拟机下改为hadoop路径同样可实现。

此为maperduce下的wordcount java代码实现。

MapReduce执行过程

数据定义格式
map: (K1,V1) → list (K2,V2)
reduce: (K2,list(V2)) → list (K3,V3)
MapReduce执行过程
Mapper
Combiner
Partitioner
Shuffle and Sort
Reducer

Spliting：先将文件分片 按照设定的hdfs块大小设定分片假如我们设定hdfs的块的大小是64mb，
如果我们输入有三个文件，大小分别是3mb、65mb和127mb，那么mapreduce会把3mb文件分为一个输入分片（input split），
65mb则是两个输入分片（input split）而127mb也是两个输入分片（input split），换句话说我们如果在map计算前做输入分片调整，
例如合并小文件，那么就会有5个map任务将执行，而且每个map执行的数据大小不均

Map：将分割的内容 再次分割成单词并且统计分割的方法是按照写好的map程序分割

Combine：Combiner是一个本地化的reduce操作，它是map运算的后续操作，
主要是在map计算出中间文件前做一个简单的合并重复key值的操作，例如我们对文件里的单词频率做统计，
map计算时候如果碰到一个hadoop的单词就会记录为1，但是这篇文章里hadoop可能会出现n多次，
那么map输出文件冗余就会很多，因此在reduce计算前对相同的key做一个合并操作，那么文件会变小，
这样就提高了宽带的传输效率，毕竟hadoop计算力宽带资源往往是计算的瓶颈也是最为宝贵的资源，但是combiner操作是有风险的，
使用它的原则是combiner的输入不会影响到reduce计算的最终输入，例如：如果计算只是求总数，最大值，最小值可以使用combiner，
但是做平均值计算使用combiner的话，最终的reduce计算结果就会出错。

Shuffle and sort:
Mapreduce 确保每个reducer的input都是按照key 排序的。系统将 map ouputs 变成 reduce inputs输入的过程被称为 shuffle。

1.Input Split分配给Map

2.Map过程进行处理，Mapper任务会接收输入分片，然后不断的调用map函数，对记录进行处理。处理完毕后，转换为新的<key,value>输出。（其中每个分片对应一个map,一个map可以被调用多次来处理该分片）

3.Map的输出结果缓存在内存里

4.内存中进行Partition，默认是HashPartitioner(采用取模hash (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks)， 目的是将map的结果分给不同的reducer，有几个Partition，就有几个reducer，partition的数目可以在job启动时通过参数 “-Dmapreduc.job.reduces”设置(Hadoop 2.2.0), HashPartitioner可以让map的结果均匀的分给不同机器上的reducer，保证负载均衡。

5.内存中在Partition结束后，对于不同分区的数据，会按照key进行排序，这里的key必须实现WritableComparable接口。该接口实现了Comparable接口，因此可以进行比较排序

6.对于排序后的<key,value>，会按照key进行分组。如果key相同，那么相同key的<key,value>就被分到一个组中。最终，每个分组会调用一次reduce函数

7.排序分组结束后，相同的key在一起组成了一个列表，如果设置过combiner，就合并数据，减少写入磁盘的记录数（combiner本质就是一个reducer）

8.当内存中buffer(default 512M)达到阈值(default 80%)，就会把记录spill(即溢写)到磁盘中，优化Map时可以调大buffer的阈值，缓存更多的数据。

9.当磁盘中的spill文件数目比规定的文件数目多时,会多次调用combiner。在不影响结果的前下,Combiner可以被调用多次。（比如磁盘中的spill文件数目规定在个（包括）以上,当达到规定数量时, map的新的output会再次运行combiner，而如果磁盘中spill file文件就1~2个，就没有必要调用combiner

10.Map结束时会把spill出来的多个文件合并成一个，merge过程最多10（默认）个文件同时merge成一个文件，多余的文件分多次merge，merge过程是merge sort的算法。

11.Map端shuffle完毕，数据都有序的存放在磁盘里，等待reducer来拿取。
Reducer端
shuffle and sort的过程不仅仅在map端，别忘了reducer端还没拿数据呢，reduce job当然不能开
启。

1. Copy phase: reducer的后台进程(default 5个)到被Application Master (Hadoop 2.2)指定的机器上将map的output拷贝到本地，先拷贝到内存，内存满了就拷贝的磁盘。
2. Sort phase(Merge phase): Reduer采用merge sort，将来自各个map的data进行merge， merge成有序的更大的文件。
3. 如果设置过Combiner，merge过程可能会调用Combiner，调不调用要看在磁盘中产生的文件数目是否超过了设定的阈值。(这一点我还没有确认，但Combiner在Reducer端是可能调用。)
4. Reduce phase: reduce job开始，输入是shuffle sort过程merge产生的文件。

Hadoop V1 MR引擎

Job Tracker

(1)概述：JobTracker是一个后台服务进程，启动之后，会一直监听并接收来自各个TaskTracker发送的心跳信息，包括资源使用情况和任务运行情况等信息。
(2)JobTracker的主要功能：
1.作业控制：在hadoop中每个应用程序被表示成一个作业，每个作业又被分成多个任务，JobTracker的作业控制模块则负责作业的分解和状态监控。
*最重要的是状态监控：主要包括TaskTracker状态监控、作业状态监控和任务状态监控。主要作用：容错和为任务调度提供决策依据。
2.资源管理。

Task Tracker

(1)TaskTracker概述：TaskTracker是JobTracker和Task之间的桥梁：一方面，从JobTracker接收并执行各种命令：运行任务、提交任务、杀死任务等；另一方面，
将本地节点上各个任务的状态通过心跳周期性汇报给JobTracker。TaskTracker与JobTracker和Task之间采用了RPC协议进行通信。
(2)TaskTracker的功能：
1.汇报心跳：Tracker周期性将所有节点上各种信息通过心跳机制汇报给JobTracker。这些信息包括两部分：
*机器级别信息：节点健康情况、资源使用情况等。
*任务级别信息：任务执行进度、任务运行状态等。
2.执行命令：JobTracker会给TaskTracker下达各种命令，主要包括：启动任务(LaunchTaskAction)、提交任务(CommitTaskAction)、杀死任务(KillTaskAction)、
杀死作业(KillJobAction)和重新初始化(TaskTrackerReinitAction)。

注意：
JobTracker 对应于 NameNode
TaskTracker 对应于 DataNode
DataNode 和NameNode 是针对数据存放来而言的
JobTracker和TaskTracker是对于MapReduce执行而言的
mapreduce中几个主要概念，mapreduce整体上可以分为这么几条执行线索：
jobclient，JobTracker与TaskTracker。
1、JobClient会在用户端通过JobClient类将应用已经配置参数打包成jar文件存储到hdfs，
并把路径提交到Jobtracker,然后由JobTracker创建每一个Task（即MapTask和ReduceTask）
并将它们分发到各个TaskTracker服务中去执行
2、JobTracker是一个master服务，软件启动之后JobTracker接收Job，负责调度Job的每一个子任务task运行于TaskTracker上，
并监控它们，如果发现有失败的task就重新运行它。一般情况应该把JobTracker部署在单独的机器上。
3、TaskTracker是运行在多个节点上的slaver服务。TaskTracker主动与JobTracker通信，接收作业，并负责直接执行每一个任务。
TaskTracker都需要运行在HDFS的DataNode上.