分布式文件系统中的MapReduce技术介绍

我们在研究分布式文件系统的实现时,不可避免要讨论MapReduce技术。比较常见的使用这一技术的有HDFS (Hadoop Distributed File System),它是Google文件系统GFS的开源实现。当然很多别的分布式文件系统,比如GlusterFS,QFS(Quantcast File System)以及一些存储服务比如Amazon的S3,Azure Blob Storage以及OpenStack Swift都或多或少有这一技术的影子。本文就来基于HDFS详细介绍一下MapReduce的技术。

分布式文件系统

所谓分布式文件系统,我们可以理解为把文件存储在多个节点中,他们之间通过网络相互连接。从而达到在客户端看起来,这些多个节点就像一台机器一样。为了达到这个效果,HDFS会有一个中心的服务器,我们称之为NameNode,它的作用就是用来知道哪台服务器中存储了哪些文件块,就像大脑一样,我们需要访问某一个文件块的时候,就可以先通过它来确定访问哪一天机器。相应的,在每一台机器上,会运行一个守护进程,它暴露了一个网络服务,这样别的节点就可以通过这个服务访问它上面存储的文件。

和数据库比较类似,为了容错,也会有replication的概念,就是把一个机器上的文件保存多个拷贝在不同的机器上。

MapReduce简介

MapReduce其实说白了是一个编程框架,你可以写相应的代码来处理大的数据集。我们来看这样一个例子:你有一个网站,然后有不同的页面,对应不同的URL,每次有人访问一个页面,就会记录一条记录。 每条记录就是URL和访问的信息,那么现在我们需要对这些记录进行分析,大概的步骤如下:

  1. 读出记录的文件,把它们分成一条条记录(比如一行表示一条记录)
  2. 调用mapper函数,从每条记录中提出一个key和value对。比如这里我们需要分析每个URL有多少人访问,可以以URL为key,value设为空即可。
  3. 根据key来进行排序
  4. 调用reducer函数来遍历所有的key-value对,并进行聚合,我们的例子中就是把所有相同的URL合并,并计算有多少条(count函数)

假如我们使用MapReduce来实现这些步骤的话,第一步就是输入格式解析,第二步就是map的操作,第四步是reduce的操作,这两部就是你需要实现相关自定义代码的部分。另外第三步的排序不需要额外来做,因为从map出来的输出就是已经排好序的。

那么第二步和第四步的map/reduce是什么意思呢?

  • Mapper:这个函数对每一个输入的记录都需要调用一次,它主要用来提取键值对。它不保存任何状态,就是对每一个记录这个操作都是独立的。
  • Reducer:处理Mapper出来的输出,对同样key的值进行处理,比如我们上面的提到的计算同样URL的数目等。

分布式执行

MapReduce最大的特点就是可以在不同的机器上并行执行,而且你不需要写什么特殊的逻辑来处理这个并行操作。

我们来看下面这个图所示的例子,它所示的就是Hadoop中的一个MapReduce的job。这里我们把输入进行partition,标志为m1,m2,m3,然后每一个partition都有一个Mapper进行处理。一般来说在决定mapper的运行的时候,会尽量选择靠近相应输入文件的机器,这样一来就可以减少拷贝输入文件的消耗。通常情况下,相应的机器其实并没有运行map的应用程序,所以会把相应的应用程序包拷贝到对应机器,然后在其上运行应用程序,并输出键值对。

在Map处理好了之后,就需要把同样的键值发送给同一个Reduce进行处理(比如图上的r1,r2,r3)等。需要注意的是我们希望进入reduce的数据是按照某种规则排序的,那么这个排序在哪里实现呢?一般来说各个Map会在它本地先进行排序,然后保存一个有序的输出在他们本地,然后通知reduce说我们这边准备好了。Reduce遍历所有的Map,拷贝相关的数据,在这个过程中再进行 一次全局的排序(merge sort),因为每个map都是排好序的,所以在reduce这一端进行排序的消耗其实是相对还蛮小的。我们把这个过程称之为Shuffle,有点confuse,因为洗牌其实是打乱顺序,而这里其实指的是重新排序。最后reduce再根据他自己的逻辑进行处理就可以了。

MapReduce Workflows

其实单个MapReduce能达到的效果是有限的,在现实中会有很多个MapReduce来组成一个链,我们称之为Workflow来处理实际问题。比如我们上面提到的统计每个URL的数目可以用一个MapReduce实现,但假如我们想知道哪些网页是Top 10访问,就需要再加一个MapReduce来实现。

需要注意的是这里的Workflow链和我们通常理解的前面的输出作为后面的输入相比有些不同,它需要把前一个MapReduce的输出保存到一个临时文件中,然后后一个MapReduce需要读这个临时文件作为新的输入。这种设计有好有坏,我们在后面再详细分析。

Reduce的join和Group

从上面的例子中我们很容易看到Reduce部分很多时候做的操作就是Join或者Group,比如计算某个URL的count,说白了就是一个count的Group操作。Reduce和数据库不太一样的是通常来说它没有index来加速这个过程,而且它需要处理的所有的输入数据,因此,一般来说使用的是全表扫描(full table scan)的技术,也就是把所有的输入文件的内容都读出来。我们来看下面这个例子:

同样是统计URL的访问数据,在记录的event中,我们使用的是userID(表示哪一个用户)来记录它的访问信息,我们有另外一张表User database来记录每一个userID的信息,比如哪一年生的,邮箱是什么之类的。这个时候我们需要统计的信息是不同年龄段所喜欢的URL的分布。

很显然这个时候,我们需要把这两张表进行join,一种简单的方法就是先把User Activity Events和User Database进行join,但是由于他们其实可能在不同的机器上,远程join的效率其实会很低,所以这里我们可以使用MapReduce来同时进行处理。如下图所示:

这里有两个map,一个产生的键值对是userid和URL,一个产生的键值对userid和生日。然后Reducer就可以先处理有序的User database Mapper,然后再处理Activity Mapper(比如按时间排序)。这种技术我们称之为Secondary sort。

这样一来,join的操作就很简单了,Reducer函数对每一个userID调用一次,第一个值就是从User database中拿到的生日,然后把生日保存到一个局部变量,再遍历Activity mapper去查找同样的UserID,这就有了一个viewed-url和viewer-age-in-year的对。有了这些信息之后,后面的MapReduce就可以很方便地计算出我们需要的东西,比如各个年龄段喜欢的URL分布等信息。

Hot Key的处理

我们上面提到的处理中,有一个同样的设置就是把同样的key发送到同一个reducer来进行处理。这种设置有一个潜在的问题,就是假如有某些key非常hot,比如说一些大V的网页会访问得很频繁,那么它的数据记录会很多,这就会导致处理这个key的reducer成为一个瓶颈,意味MapReduce有一个条件就是所有的reducer都结束了才算这个job真的结束。那么一个reducer成为瓶颈就意味着整个job都需要等待它完成。

为了解决这个问题,业界也有很多相应的算法,一个常见的算法就是首先确认哪个key是hot key,然后在做join的时候,让这个key随机发送给多个reducer进行处理。这样就可以分散reducer的处理压力到多个节点,只是有个问题就是后期再进行key的处理的时候,你需要把所有和hot key相关的reduce都再次进行重复发送才行。

为了解决重复发送这个问题,一个新的算法是在hot key选择处理reduce节点的时候不再水随机选择,而是根据一定规则进行处理,这样一来的好处就是再进行处理的时候我们可以知道哪些reducer中的hot key是需要的,就不需要每次都进行重复发送了。

总结

至此我们就通过一些实例介绍了MapReduce相关的概念和可能遇到的问题,希望能够对大家理解MapReduce有所帮助。

You may also like...

Leave a Reply

Your email address will not be published. Required fields are marked *