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

问题陈述：

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

解决方案：

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

 
     
 
      1 
      class N 
      2 State 
      in {True = 
      2 , False = 
      1 , 
      null = 
      0 }, 
      3 initialized 
      1 or 
      2 
      for end - of - line categories, 
      0 otherwise 
      4 method getMessage( 
      object N) 
      5 
      return N.State 
      6 method calculateState(state s, data [d1, d2,...]) 
      7 
      return max( [d1, d2,...] ) 
     
 
    
 
    
案例研究：广度优先搜索
 
    
问题陈述：需要计算出一个图结构中某一个节点到其它所有节点的距离。
 
    
解决方案： Source源节点给所有邻接点发出值为0的信号，邻接点把收到的信号再转发给自己的邻接点，每转发一次就对信号值加1：
 
    
 
     
 
     
 
      1 
      class N 
      2 State 
      is distance, 
      3 initialized 
      0 
      for source node, INFINITY 
      for all other nodes 
      4 method getMessage(N) 
      5 
      return N.State + 
      1 
      6 method calculateState(state s, data [d1, d2,...]) 
      7 min( [d1, d2,...] ) 
     
 
    
 
    
案例研究：网页排名和 Mapper 端数据聚合
 
    
这个算法由Google提出，使用权威的PageRank算法，通过连接到一个网页的其他网页来计算网页的相关性。真实算法是相当复杂的，但是核心思想是权重可以传播，也即通过一个节点的各联接节点的权重的均值来计算节点自身的权重。
 
    
 
     
 
     
 
      1 
      class N 
      2 State 
      is PageRank 
      3 method getMessage( 
      object N) 
      4 
      return N.State / N.OutgoingRelations.size() 
      5 method calculateState(state s, data [d1, d2,...]) 
      6 
      return ( sum([d1, d2,...]) ) 
     
 
    
 
    
要指出的是上面用一个数值来作为评分实际上是一种简化，在实际情况下，我们需要在Mapper端来进行聚合计算得出这个值。下面的代码片段展示了这个改变后的逻辑 （针对于 PageRank 算法）：
 
    
 
     
 
     
 
       1 
      class Mapper 
       2 method Initialize 
       3 H = 
      new AssociativeArray 
       4 method Map(id n, 
      object N) 
       5 p = N.PageRank / N.OutgoingRelations.size() 
       6 Emit(id n, 
      object N) 
       7 
      for all id m 
      in N.OutgoingRelations 
      do 
       8 H{m} = H{m} + p 
       9 method Close 
      10 
      for all id n 
      in H 
      do 
      11 Emit(id n, value H{n}) 
      12 
      13 
      class Reducer 
      14 method Reduce(id m, [s1, s2,...]) 
      15 M = 
      null 
      16 p = 
      0 
      17 
      for all s 
      in [s1, s2,...] 
      do 
      18 
      if IsObject(s) then 
      19 M = s 
      20 
      else 
      21 p = p + s 
      22 M.PageRank = p 
      23 Emit(id m, item M) 
     
 
    
 
    
应用：图分析，网页索引
 
    
 
 
    
 
 
    
值去重 （对唯一项计数）
 
    
问题陈述: 记录包含值域F和值域 G，要分别统计相同G值的记录中不同的F值的数目 (相当于按照 G分组).
 
    
这个问题可以推而广之应用于分面搜索（某些电子商务网站称之为Narrow Search）
 
    
 
     
Record 1: F=1, G={a, b}  Record 2: F=2, G={a, d, e}  Record 3: F=1, G={b}  Record 4: F=3, G={a, b}  Result:  a -> 3 // F=1, F=2, F=3  b -> 2 // F=1, F=3  d -> 1 // F=2  e -> 1 // F=2
 
    
 
    
解决方案 I:
 
    
第一种方法是分两个阶段来解决这个问题。第一阶段在Mapper中使用F和G组成一个复合值对，然后在Reducer中输出每个值对，目的是为了保证F值的唯一性。在第二阶段，再将值对按照G值来分组计算每组中的条目数。
 
    
第一阶段：
 
    
 
     
 
     
 
      1 
      class Mapper 
      2 method Map( 
      null , record [value f, categories [g1, g2,...]]) 
      3 
      for all category g 
      in [g1, g2,...] 
      4 Emit(record [g, f], count 
      1 ) 
      5 
      6 
      class Reducer 
      7 method Reduce(record [g, f], counts [n1, n2, ...]) 
      8 Emit(record [g, f], 
      null ) 
     
 
    
 
    
第二阶段：
 
    
 
     
 
     
 
      1 
      class Mapper 
      2 method Map(record [f, g], 
      null ) 
      3 Emit(value g, count 
      1 ) 
      4 
      5 
      class Reducer 
      6 method Reduce(value g, counts [n1, n2,...]) 
      7 Emit(value g, sum( [n1, n2,...] ) ) 
     
 
    
 
    
解决方案 II:
 
    
第二种方法只需要一次MapReduce 即可实现，但扩展性不强。算法很简单-Mapper 输出值和分类，在Reducer里为每个值对应的分类去重然后给每个所属的分类计数加1，最后再在Reducer结束后将所有计数加和。这种方法适用于只有有限个分类，而且拥有相同F值的记录不是很多的情况。例如网络日志处理和用户分类，用户的总数很多，但是每个用户的事件是有限的，以此分类得到的类别也是有限的。值得一提的是在这种模式下可以在数据传输到Reducer之前使用Combiner来去除分类的重复值。
 
    
 
     
 
     
 
       1 
      class Mapper 
       2 method Map( 
      null , record [value f, categories [g1, g2,...] ) 
       3 
      for all category g 
      in [g1, g2,...] 
       4 Emit(value f, category g) 
       5 
       6 
      class Reducer 
       7 method Initialize 
       8 H = 
      new AssociativeArray : category -> count 
       9 method Reduce(value f, categories [g1, g2,...]) 
      10 [g1 
      ' 
      , g2 
      ' ,..] = ExcludeDuplicates( [g1, g2,..] ) 
      11 
      for all category g 
      in [g1 
      ' 
      , g2 
      ' ,...] 
      12 H{g} = H{g} + 
      1 
      13 method Close 
      14 
      for all category g 
      in H 
      do 
      15 Emit(category g, count H{g}) 
     
 
    
 
    
应用：日志分析，用户计数
 
    
 
 
    
互相关
 
    
问题陈述：有多个各由若干项构成的组，计算项两两共同出现于一个组中的次数。假如项数是N，那么应该计算N*N。
 
    
这种情况常见于文本分析（条目是单词而元组是句子），市场分析（购买了此物的客户还可能购买什么）。如果N*N小到可以容纳于一台机器的内存，实现起来就比较简单了。
 
    
配对法
 
    
第一种方法是在Mapper中给所有条目配对，然后在Reducer中将同一条目对的计数加和。但这种做法也有缺点：
 
    
     
         
  • 使用 combiners 带来的的好处有限，因为很可能所有项对都是唯一的
    •  
         
  • 不能有效利用内存
    •  
        
 
    
 
     
 
     
 
       1 
      class Mapper 
       2 method Map( 
      null , items [i1, i2,...] ) 
       3 
      for all item i 
      in [i1, i2,...] 
       4 
      for all item j 
      in [i1, i2,...] 
       5 Emit(pair [i j], count 
      1 ) 
       6 
       7 
      class Reducer 
       8 method Reduce(pair [i j], counts [c1, c2,...]) 
       9 s = sum([c1, c2,...]) 
      10 Emit(pair[i j], count s) 
     
 
    
 
    
Stripes Approach（条方法？不知道这个名字怎么理解）
 
    
第二种方法是将数据按照pair中的第一项来分组，并维护一个关联数组，数组中存储的是所有关联项的计数。The second approach is to group data by the first item in pair and maintain an associative array (“stripe”) where counters for all adjacent items are accumulated. Reducer receives all stripes for leading item i, merges them, and emits the same result as in the Pairs approach.
 
    
     
         
  • 中间结果的键数量相对较少，因此减少了排序消耗。
    •  
         
  • 可以有效利用 combiners。
    •  
         
  • 可在内存中执行，不过如果没有正确执行的话也会带来问题。
    •  
         
  • 实现起来比较复杂。
    •  
         
  • 一般来说， “stripes” 比 “pairs” 更快
    •  
        
 
    
 
     
 
     
 
       1 
      class Mapper 
       2 method Map( 
      null , items [i1, i2,...] ) 
       3 
      for all item i 
      in [i1, i2,...] 
       4 H = 
      new AssociativeArray : item -> counter 
       5 
      for all item j 
      in [i1, i2,...] 
       6 H{j} = H{j} + 
      1 
       7 Emit(item i, stripe H) 
       8 
       9 
      class Reducer 
      10 method Reduce(item i, stripes [H1, H2,...]) 
      11 H = 
      new AssociativeArray : item -> counter 
      12 H = merge - sum( [H1, H2,...] ) 
      13 
      for all item j 
      in H.keys() 
      14 Emit(pair [i j], H{j}) 
     
 
    
 
    
应用：文本分析，市场分析
 
    
参考资料：Lin J. Dyer C. Hirst G. 
 
    
 
 
    
用MapReduce 表达关系模式
 
    
在这部分我们会讨论一下怎么使用MapReduce来进行主要的关系操作。
 
    
筛选（Selection）
 
    
 
     
 
     
 
      1 
      class Mapper 
      2 method Map(rowkey key, tuple t) 
      3 
      if t satisfies the predicate 
      4 Emit(tuple t, 
      null ) 
     
 
    
 
    
投影（Projection）
 
    
投影只比筛选稍微复杂一点，在这种情况下我们可以用Reducer来消除可能的重复值。
 
    
 
     
 
     
 
      1 
      class Mapper 
      2 method Map(rowkey key, tuple t) 
      3 tuple g = project(t) 
      // 
       extract required fields to tuple g 
        
      4 Emit(tuple g, 
      null ) 
      5 
      6 
      class Reducer 
      7 method Reduce(tuple t, array n) 
      // 
       n is an array of nulls 
        
      8 Emit(tuple t, 
      null ) 
     
 
    
 
    
合并（Union）
 
    
两个数据集中的所有记录都送入Mapper，在Reducer里消重。
 
    
 
     
 
     
 
      1 
      class Mapper 
      2 method Map(rowkey key, tuple t) 
      3 Emit(tuple t, 
      null ) 
      4 
      5 
      class Reducer 
      6 method Reduce(tuple t, array n) 
      // 
       n is an array of one or two nulls 
        
      7 Emit(tuple t, 
      null ) 
     
 
    
 
    
交集（Intersection）
 
    
将两个数据集中需要做交叉的记录输入Mapper，Reducer 输出出现了两次的记录。因为每条记录都有一个主键，在每个数据集中只会出现一次，所以这样做是可行的。
 
    
 
     
 
     
 
      1 
      class Mapper 
      2 method Map(rowkey key, tuple t) 
      3 Emit(tuple t, 
      null ) 
      4 
      5 
      class Reducer 
      6 method Reduce(tuple t, array n) 
      // 
       n is an array of one or two nulls 
        
      7 
      if n.size() = 
      2 
      8 Emit(tuple t, 
      null ) 
     
 
    
 
    
差异（Difference）
 
    
假设有两个数据集R和S，我们要找出R与S的差异。Mapper将所有的元组做上标记，表明他们来自于R还是S，Reducer只输出那些存在于R中而不在S中的记录。
 
    
 
     
 
     
 
      1 
      class Mapper 
      2 method Map(rowkey key, tuple t) 
      3 Emit(tuple t, 
      string t.SetName) 
      // 
       t.SetName is either 'R' or 'S' 
        
      4 
      5 
      class Reducer 
      6 method Reduce(tuple t, array n) 
      // 
       array n can be ['R'], ['S'], ['R' 'S'], or ['S', 'R'] 
        
      7 
      if n.size() = 
      1 and n[ 
      1 ] = 
      ' 
      R 
      ' 
      8 Emit(tuple t, 
      null ) 
     
 
    
 
    
分组聚合（GroupBy and Aggregation）
 
    
分组聚合可以在如下的一个MapReduce中完成。Mapper抽取数据并将之分组聚合，Reducer 中对收到的数据再次聚合。典型的聚合应用比如求和与最值可以以流的方式进行计算，因而不需要同时保有所有的值。但是另外一些情景就必须要两阶段MapReduce，前面提到过的惟一值模式就是一个这种类型的例子。
 
    
 
     
 
     
 
      1 
      class Mapper 
      2 method Map( 
      null , tuple [value GroupBy, value AggregateBy, value ...]) 
      3 Emit(value GroupBy, value AggregateBy) 
      4 
      5 
      class Reducer 
      6 method Reduce(value GroupBy, [v1, v2,...]) 
      7 Emit(value GroupBy, aggregate( [v1, v2,...] ) ) 
      8 
      // 
       aggregate() : sum(), max(),... 
     
 
    
 
    
连接（Joining）
 
    
MapperReduce框架可以很好地处理连接，不过在面对不同的数据量和处理效率要求的时候还是有一些技巧。在这部分我们会介绍一些基本方法，在后面的参考文档中还列出了一些关于这方面的专题文章。
 
    
分配后连接 （Reduce端连接,排序-合并连接）
 
    
这个算法按照键K来连接数据集R和L。Mapper 遍历R和L中的所有元组，以K为键输出每一个标记了来自于R还是L的元组，Reducer把同一个K的数据分装入两个容器（R和L），然后嵌套循环遍历两个容器中的数据以得到交集，最后输出的每一条结果都包含了R中的数据、L中的数据和K。这种方法有以下缺点：
 
    
     
         
  • Mapper要输出所有的数据，即使一些key只会在一个集合中出现。
    •  
         
  • Reducer 要在内存中保有一个key的所有数据，如果数据量打过了内存，那么就要缓存到硬盘上，这就增加了硬盘IO的消耗。
    •  
        
 
    
尽管如此，再分配连接方式仍然是最通用的方法，特别是其他优化技术都不适用的时候。
 
    
 
     
 
     
 
       1 
      class Mapper 
       2 method Map( 
      null , tuple [join_key k, value v1, value v2,...]) 
       3 Emit(join_key k, tagged_tuple [set_name tag, values [v1, v2, ...] ] ) 
       4 
       5 
      class Reducer 
       6 method Reduce(join_key k, tagged_tuples [t1, t2,...]) 
       7 H = 
      new AssociativeArray : set_name -> values 
       8 
      for all tagged_tuple t 
      in [t1, t2,...] 
      // 
       separate values into 2 arrays 
        
       9 H{t.tag}.add(t.values) 
      10 
      for all values r 
      in H{ 
       
      ' 
      R 
      ' } 
      // 
       produce a cross-join of the two arrays 
        
      11 
      for all values l 
      in H{ 
       
      ' 
      L 
      ' } 
      12 Emit( 
      null , [k r l] ) 
     
 
    
 
    
复制链接Replicated Join （Mapper端连接, Hash 连接）
 
    
在实际应用中，将一个小数据集和一个大数据集连接是很常见的（如用户与日志记录）。假定要连接两个集合R和L，其中R相对较小，这样，可以把R分发给所有的Mapper，每个Mapper都可以载入它并以连接键来索引其中的数据，最常用和有效的索引技术就是哈希表。之后，Mapper遍历L，并将其与存储在哈希表中的R中的相应记录连接，。这种方法非常高效，因为不需要对L中的数据排序，也不需要通过网络传送L中的数据，但是R必须足够小到能够分发给所有的Mapper。
 
    
 
     
 
     
 
       1 
      class Mapper 
       2 method Initialize 
       3 H = 
      new AssociativeArray : join_key -> tuple from R 
       4 R = loadR() 
       5 
      for all [ join_key k, tuple [r1, r2,...] ] 
      in R 
       6 H{k} = H{k}.append( [r1, r2,...] ) 
       7 
       8 method Map(join_key k, tuple l) 
       9 
      for all tuple r 
      in H{k} 
      10 Emit( 
      null , tuple [k r l] ) 
     
 
    
 
    
参考资料：
 
    
     
         
  1.  
         
  2.  
        
 
    
应用于机器学习和数学方面的 MapReduce 算法
 
    
     
         
  • C. T. Chu et al provides an excellent description of  machine learning algorithms for MapReduce in the article .
    •  
         
  • FFT using MapReduce:  
    •  
         
  • MapReduce for integer factorization:
    •  
         
  • Matrix multiplication with MapReduce:  and
    •