- 微信
- 微博
  
  分享文章到微博
- 复制链接
  
  复制链接到剪贴板

MapReduce编程模型和计算框架

JavaEdge 发表于 2022/04/05 20:53:18 2022/04/05

【摘要】 1 概述源于Google的MapReduce论文，发表于2004年12月。Hadoop MapReduce是Google MapReduce的克隆版。Hadoop问世前，已有分布式计算，但都是专用系统，仅处理某一类计算，比如进行大规模数据排序。这样的系统无法复用到其他大数据计算场景，每种应用都需要开发与维护专门系统。而Hadoop MapReduce造就了大数据计算通用编程。只要遵循Map...

1 概述

源于Google的MapReduce论文，发表于2004年12月。Hadoop MapReduce是Google MapReduce的克隆版。Hadoop问世前，已有分布式计算，但都是专用系统，仅处理某一类计算，比如进行大规模数据排序。这样的系统无法复用到其他大数据计算场景，每种应用都需要开发与维护专门系统。

而Hadoop MapReduce造就了大数据计算通用编程。只要遵循MapReduce编程模型编写业务处理逻辑代码，就能运行在Hadoop分布式集群。我们只需关心业务逻辑，无需关心系统调用与运行环境。

大数据计算的核心思路：移动计算比移动数据划算。既然计算方法跟传统计算方法不同，移动计算而非移动数据，那用传统编程模型进行大数据计算就会遇到很多困难，因此Hadoop使用MapReduce的编程模型。

MapReduce编程模型并非Hadoop原创，也非Google原创，但Google和Hadoop创造性地将MapReduce编程模型用到大数据计算，让复杂的各种各样的机器学习、数据挖掘、SQL处理等大数据计算变得简单。

Hadoop解决大规模数据分布式计算的方案——MapReduce。

MapReduce既是编程模型，也是计算框架：开发人员必须基于MapReduce编程模型进行code，然后将程序通过MapReduce计算框架分发到Hadoop集群中运行。

MapReduce编程模型执行步骤

准备map处理的输入数据
Mapper处理
Shuffle
Reduce处理
结果输出
InputFormat

OutputFormat

决定在哪里以及怎样持久化作业结果。Hadoop为不同类型的格式提供了一系列的类和接口，实现自定义操作只要继承其中的某个类或接口即可。你可能已经熟悉了默认的OutputFormat，也就是TextOutputFormat，它是一种以行分隔，包含制表符界定的键值对的文本文件格式。尽管如此，对多数类型的数据而言，如再常见不过的数字，文本序列化会浪费一些空间，由此带来的结果是运行时间更长且资源消耗更多。为了避免文本文件的弊端，Hadoop提供了SequenceFileOutputformat，它将对象表示成二进制形式而不再是文本文件，并将结果进行压缩。

Split

Split：交由MapReduce作业来处理的数据块，是MapReduce中最小的计算单元。
HDFS：blocksize是HDFS中最小的存储单元，128M
默认他们两一一对应，当然我们也可以手工设置他们之间的关系(不建议)

2 MapReduce编程模型

模型只包含Map、Reduce两个过程：

map的输入是一对<Key, Value>值，经过map计算后输出一对<Key, Value>值；然后将相同Key合并，形成<Key, Value集合>
再将这个<Key, Value集合>输入reduce，经过计算输出零个或多个<Key, Value>对

强大，不管是关系代数运算（SQL计算），还是矩阵运算（图计算），大数据领域几乎所有的计算需求都可以通过MapReduce实现。

3 WordCount

文本处理中词频统计的问题，就是统计文本中每一个单词出现的次数。如果只是统计一篇文章的词频，几十KB到几MB的数据，只需要写一个程序，将数据读入内存，建一个Hash表记录每个词出现的次数就可以了。这个统计过程你可以看下面这张图。

3.1 单机处理

# 文本前期处理
strl_ist = str.replace('\n', '').lower().split(' ')
count_dict = {}
# 如果字典里有该单词则加1，否则添加入字典
for str in strl_ist:
if str in count_dict.keys():
    count_dict[str] = count_dict[str] + 1
    else:
        count_dict[str] = 1

建个Hash表，然后将字符串里的每个词放进Hash表。若该词第一次放到Hash表，就新建一个KV对，K=该词，V=1。若Hash表里已有该词，则给该词的V + 1。

小数据量用单机统计词频很简单，但若想统计互联网所有网页（万亿计）的词频数（Google这种体量搜索引擎的需求），不可能写一个程序把全世界的网页都读入内存，这时候就需要用MapReduce编程。

3.2 WordCount的MapReduce

public class WordCount {

  public static class TokenizerMapper
       extends Mapper<Object, Text, Text, IntWritable>{

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }

  public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }
}

MapReduce版本WordCount程序的核心是map、reduce函数。

map函数的输入主要是一个<Key, Value>对，在这个例子里，Value是要统计的所有文本中的一行数据，Key在一般计算中都不会用到。

public void map(Object key, Text value, Context context
                    )

map函数的计算过程是，将这行文本中的单词提取出来，针对每个单词输出一个<word, 1>这样的KV对。

MapReduce计算框架会将这些<word , 1>收集起来，将相同的word放在一起，形成<word , <1,1,1,1,1,1,1…>>这样的<Key, Value集合>数据，然后将其输入给reduce函数。

public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       )

这里reduce的输入参数Values就是由很多个1组成的集合，而Key就是具体的单词word。

reduce将这个集合里的1求和，再将单词（word）和这个和（sum）组成一个<Key, Value>，即<word, sum>输出，即一个单词和它的词频统计总和。

一个map函数能针对一部分数据进行运算，这就能将一个大数据切分成很多块（HDFS），MapReduce计算框架为每个数据块分配一个map函数去计算，从而实现大数据的分布式计算。

假设有两个数据块的文本数据需要进行词频统计，MapReduce计算过程：

但这样一个MapReduce程序要想在分布式环境中执行，并处理海量的大规模数据，还需一个计算框架，调度执行这个MapReduce程序，使它在分布式的集群中并行运行，这个计算框架也叫MapReduce。

所以，说MapReduce时：

可能指编程模型，即一个MapReduce程序
也可能是指计算框架，调度执行大数据的分布式计算

这个过程有两个关键问题：

如何为每个数据块分配一个Map计算任务，即代码如何发送到数据块所在服务器，发送后如何启动，启动后如何知道自己需要计算的数据在文件什么位置（BlockID是啥）
处于不同服务器的map输出的<Key, Value> ，如何把相同的Key聚合在一起发送给Reduce任务进行处理

这俩问题对应MapReduce计算过程图中的两处“MapReduce框架处理”：

MapReduce作业启动和运行
MapReduce数据合并与连接

4 MapReduce作业启动和运行机制

以Hadoop 1为例，MapReduce运行过程涉及三类关键进程：

4.1 大数据应用进程

启动MapReduce程序的主入口。指定Map和Reduce类、输入输出文件路径等，并提交作业给Hadoop集群，即JobTracker进程。这是由用户启动的MapReduce程序进程，如WordCount程序。

4.2 JobTracker进程

根据要处理的输入数据量，命令TaskTracker进程启动相应数量的Map和Reduce进程任务，并管理整个作业生命周期的任务调度和监控。

这是Hadoop集群的常驻进程，JobTracker进程在整个Hadoop集群全局唯一。

4.3 TaskTracker进程

负责启动和管理Map进程以及Reduce进程。因为需要每个数据块都有对应的map函数，TaskTracker进程通常和HDFS的DataNode进程启动在同一个服务器。即Hadoop集群中绝大多数服务器同时运行DataNode进程和TaskTracker进程。

JobTracker进程和TaskTracker进程是主从关系：

主服务器通常只有一台或另有一台备机提供高可用，但运行时只有一台服务器对外服务，真正起作用的只有一台
从服务器可能有几百上千台，所有从服务器听从主服务器的控制和调度安排。主服务器负责为应用程序分配服务器资源以及作业执行的调度，具体计算操作在从服务器完成

MapReduce主服务器JobTracker，从服务器TaskTracker。HDFS主服务器NameNode，从服务器DataNode，Yarn、Spark等都是类似的一主多从的服务器架构：主服务器一台，掌控全局；从服务器多台，负责具体事情。这样很多台服务器可以有效组织起来，对外表现出一个统一又强大的计算能力。

5 作业启动和计算过程

1、应用进程JobClient将用户作业jar包存储在HDFS，这些JAR包后续会分发给Hadoop集群中的服务器执行MapReduce计算

2、应用程序提交job作业给JobTracker

3、JobTracker根据作业调度策略创建JobInProcess树，每个作业都会有一个自己的JobInProcess树

4、JobInProcess根据输入数据分片数目（一般就是数据块的数目）和设置的Reduce数目创建相应数量的TaskInProcess

5、TaskTracker进程和JobTracker进程定时通信

6、若TaskTracker有空闲的计算资源（有空闲CPU核心），JobTracker就会给它分配任务。分配任务时会根据TaskTracker的服务器名字匹配在同一台机器上的数据块计算任务给它，使启动的计算任务正好处理本机上的数据，实现“移动计算比移动数据划算”

7、TaskTracker收到任务后根据任务类型（Map or Reduce）和任务参数（作业jar包路径、输入数据文件路径、要处理的数据在文件中的起始位置和偏移量、数据块多个备份的DataNode主机名等），启动相应的Map或Reduce进程。

8、Map或者Reduce进程启动后，检查本地是否有要执行任务的JAR包文件，若无，就去HDFS下载，然后加载Map或Reduce代码开始执行。

9、若为Map进程，从HDFS读数据（通常要读取的数据块正好存储在本机）；若为Reduce进程，将结果数据写到HDFS

这样，MapReduce能将大数据作业计算任务分布在整个Hadoop集群中运行，每个Map计算任务要处理的数据通常都能从本地磁盘读取到。

而我们要做的仅是编写一个map函数和一个reduce函数，其余一切都由MapReduce计算框架自动完成！

6 MapReduce数据合并与连接机制

WordCount想统计相同单词在所有输入数据中出现的次数，而一个Map只能处理一部分数据，一个热门单词几乎会出现在所有Map中，这意味着同一个单词必须要合并到一起进行统计才能得到正确结果。

几乎所有的大数据计算场景都需要处理数据关联，像WordCount这种简单的只要对Key进行合并，对于像数据库的join操作这种复杂的，需要对两种类型（或者更多类型）的数据根据Key进行连接。

在map输出与reduce输入之间，MapReduce计算框架处理数据合并与连接操作-shuffle

7 shuffle

每个Map任务的计算结果都会写入本地文件系统，等Map任务快计算完成时，MapReduce计算框架会启动shuffle过程，在Map任务进程调用一个Partitioner接口，对Map产生的每个KV进行Reduce分区选择，然后通过HTTP通信发送给对应的Reduce进程。

这样不管Map位于哪个服务器节点，相同K一定会被发到相同Reduce进程。Reduce任务进程对收到的KV进行排序和合并，相同K放一起，组成一个<Key, Value集合>传递给Reduce执行。

map输出的<Key, Value>shuffle到哪个Reduce进程是关键，由Partitioner实现，MapReduce框架默认的Partitioner用Key的哈希值对Reduce任务数量取模，相同K落在相同Reduce任务ID：

 /** Use {@link Object#hashCode()} to partition. */ 
public int getPartition(K2 key, V2 value, int numReduceTasks) { 
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks; 
 }

分布式计算需要将不同服务器上的相关数据合并到一起进行下一步计算，这就是shuffle。

只要是大数据批处理计算，一定有shuffle过程，只有让数据关联起来，数据的内在关系和价值才会呈现出来。若不理解shuffle，在map和reduce编程中肯定困惑，如何正确设计map的输出和reduce的输入。shuffle也是整个MapReduce过程中最难、最消耗性能，MapReduce早期的一半代码都是shuffle处理。