- 详解MapReduce的模式、算法和用例

#e#

非基本 MapReduce 模式

迭代消息传递 （图处理）

问题陈述：

假设一个实体网络，实体之间存在着关系。 需要按照与它比邻的其他实体的属性计算出一个状态。这个状态可以表现为它和其它节点之间的距离， 存在特定属性的邻接点的迹象， 邻域密度特征等等。

解决方案：

网络存储为系列节点的结合，每个节点包含有其所有邻接点ID的列表。按照这个概念，MapReduce 迭代进行，每次迭代中每个节点都发消息给它的邻接点。邻接点根据接收到的信息更新自己的状态。当满足了某些条件的时候迭代停止，如达到了最大迭代次数（网络半径）或两次连续的迭代几乎没有状态改变。从技术上来看，Mapper 以每个邻接点的ID为键发出信息，所有的信息都会按照接受节点分组，reducer 就能够重算各节点的状态然后更新那些状态改变了的节点。下面展示了这个算法：

class Mapper

method Map（id n， object N）

Emit（id n， object N）

for all id m in N.OutgoingRelations do

Emit（id m， message getMessage（N））

class Reducer

method Reduce（id m， ［s1， s2，。..］）

M = null

messages = ［］

for all s in ［s1， s2，。..］ do

if IsObject（s） then

M = s

else // s is a message

messages.add（s）

M.State = calculateState（messages）

Emit（id m， item M）

一个节点的状态可以迅速的沿着网络传全网，那些被感染了的节点又去感染它们的邻居，整个过程就像下面的图示一样：

 

案例研究： 沿分类树的有效性传递

问题陈述：

这个问题来自于真实的电子商务应用。将各种货物分类，这些类别可以组成一个树形结构，比较大的分类（像男人、女人、儿童）可以再分出小分类（像男裤或女装），直到不能再分为止（像男式蓝色牛仔裤）。这些不能再分的基层类别可以是有效（这个类别包含有货品）或者已无效的（没有属于这个分类的货品）。如果一个分类至少含有一个有效的子分类那么认为这个分类也是有效的。我们需要在已知一些基层分类有效的情况下找出分类树上所有有效的分类。

解决方案：

这个问题可以用上一节提到的框架来解决。我们咋下面定义了名为 getMessage和 calculateState 的方法：

class N

State in {True = 2， False = 1， null = 0}，

initialized 1 or 2 for end-of-line categories， 0 otherwise

method getMessage（object N）

return N.State

method calculateState（state s， data ［d1， d2，。..］）

return max（ ［d1， d2，。..］ ）

案例研究：广度优先搜索

问题陈述：需要计算出一个图结构中某一个节点到其它所有节点的距离。

解决方案： Source源节点给所有邻接点发出值为0的信号，邻接点把收到的信号再转发给自己的邻接点，每转发一次就对信号值加1：

class N

State is distance，

initialized 0 for source node， INFINITY for all other nodes

method getMessage（N）

return N.State + 1

method calculateState（state s， data ［d1， d2，。..］）

min（ ［d1， d2，。..］ ）

案例研究：网页排名和 Mapper 端数据聚合

这个算法由Google提出，使用权威的PageRank算法，通过连接到一个网页的其他网页来计算网页的相关性。真实算法是相当复杂的，但是核心思想是权重可以传播，也即通过一个节点的各联接节点的权重的均值来计算节点自身的权重。

class N

State is PageRank

method getMessage（object N）

return N.State / N.OutgoingRelations.size（）

method calculateState（state s， data ［d1， d2，。..］）

return （ sum（［d1， d2，。..］） ）

要指出的是上面用一个数值来作为评分实际上是一种简化，在实际情况下，我们需要在Mapper端来进行聚合计算得出这个值。下面的代码片段展示了这个改变后的逻辑 （针对于 PageRank 算法）：

class Mapper

method Initialize

H = new AssociativeArray

method Map（id n， object N）

p = N.PageRank / N.OutgoingRelations.size（）

Emit（id n， object N）

for all id m in N.OutgoingRelations do

H{m} = H{m} + p

method Close

for all id n in H do

Emit（id n， value H{n}）

class Reducer

method Reduce（id m， ［s1， s2，。..］）

M = null

p = 0

for all s in ［s1， s2，。..］ do

if IsObject（s） then

M = s

else

p = p + s

M.PageRank = p

Emit（id m， item M）

应用：

图分析，网页索引