MapReduce设计模式：Filtering

MapReduce设计模式：Filtering

 

在上一节中，我们学习了MapReduce的设计模式之一：Local Aggregation；这个更像是对MapReduce的优化，确保数据能够准确、快速、高效的运行。本节学习的Filtering模式则是从另外一个角度来看数据。

 

在Filtering模式下，我们不是想改动源数据，只是想得到源数据的子集；该集合有可能很小，如TOP-K；也有可能很大，如数据去重。使用Filtering时，你肯定是很清楚业务功能，想从更高的层面去看待数据：从语料库里抽取最常用词、StackOverflow用户的访问轨迹、数据的代表性子集合等，这些都是从数据中抽取部分内容，从而能够更近的观察数据，这些在机器学习和数据挖掘中都是很常见的内容。

 

我们看下常用的Filtering使用案例：

事件跟踪：更具用户的IP、ID等来分析用户感兴趣的商品、感兴趣的话题、想了解的内容，以便在用户下次登录时能够向用户做内容推荐。

内容查询：分布式查找（Distributed grep）就是一个典型的使用案例

数据清洗：数据格式、完整性、内容缺失等，导致数据不可用，需要将这些数据过滤掉，在数据挖掘中是数据的很重要的预处理过程。

抽样：这个和概率上的抽样含义一致，就是想得到子集合。有人会有疑问，数据越多越好，为什么要抽样呢？处理百万级别的数据和处理亿万级别的数据，结果差别不到的话，为什么不选择前者呢？千万别问我为什么会存在这种场景。

去重：根据用户的ID去重，得到网站的UV统计

 

使用的Filtering方法需要根据业务来判断：可以使用Regex Filtering、当然也可以直接使用Probability Sampling、或者使用特定功能过滤的Bloom Filtering；

 

 

我们先看下Regex匹配的MapReduce程序，这个实际上就是Distributed Grep的实例：

class Mapper
  method setup
    regex = new Regex Array
  method map(value val)
    if(val matches regex)
      emit(null,val)

 

这个分布式的Grep操作看起来很是简单，实际上也确实是简单；至少Hadoop提供的教程里面也是这么写的，当然写的有点不一样，我这里只关心能体现Filtering逻辑的内容。注意这个Distributed Grep没有Combiner、没有Reducer、只有一个Mapper就足够了，设置Reducer数目为Zero就不会运行Reducer了。

 

在使用概率抽样时，SRS（Simple Random Sampling）是个不错的方法。对于减小数据集大小的抽样，使用SRS方法先设定好取样的数据集合大小门限theta，然后对于每条数据，使用随机数产生数字alpha，如果alpha小于theta，该数据满足取样要求，否则丢弃。伪码如下：

class Mapper
  method setup
    theta = threshold from configuration
  method map(value val)
    alpha = new random
    if(alpha < theta)
      emit(null,val)

 

说到这些Filter，必须得提下Bloom Filter，是一种Space Efficient的概率型数据结构（probabilistic data structure），用于判定一个元素是否位于集合中。在垃圾邮件过滤的黑白名单方法、爬虫的网址判重模块中经常被用到。wiki可以参考这里（http://en.wikipedia.org/wiki/Bloom_filter）。简单说就是利用多个Hash函数来判定是否位于集合中，False Positive可能发生，就是误判；但是False Negative绝对不会发生。wiki页面上有比较详细的推导，实际上就是使用错误率来换取存储空间。如果你需要的场景是对误判的Zero-Tolerance，这个Bloom Filter不是你的选择。

使用多个Hash函数来处理碰撞，那么高效的Hash函数就是个关键，在Hadoop和Hbase中使用了MurmurHash( http://en.wikipedia.org/wiki/MurmurHash) ，Google构建于其上的CityHash (http://en.wikipedia.org/wiki/CityHash) 都具有不错的性能，原理大家自行细读。

 

明白了Bloom Filter的原理，对于使用Bloom Filter的场景也要有清晰的认识。

过滤非热词：从数据集合中过滤热词，根据Bloom Filter的特性，非热词也有可能出现，但是热词都已经存在，非热词作为噪声处理也没有关系，至少数据集合大小已经降下来了。

预处理集合：这个预处理需要仔细定义一下，假如说某个查找功能非常耗时耗力耗资源，使用这个功能时，如果能对被查找的集合做Bloom Filter处理，能够大大降低被查找的集合大小，这个预处理功能就显得很有必要了。

空间操作：对于及其耗时的硬盘操作，可以使用Bloom Filter来处理，如Google Bigtable disk lookup、Squid Web proxy cache digests等

 

一般来说，使用Bloom Filter都有个初始化的过程，会初始化Bloom Filter的Hash函数，Bit大小等，使用Bloom Filter的伪码如下：

class Mapper
  method setup
    filter = new bloom filter
  method map(value val)
    if(filter.test(val))
      emit(null,val)

 

使用Bloom Filter的内容很简单，但是使用Bloom Filter的场景判定很关键。

 

TOP-K算法在网络上有许多，也有很多的分析，其实TOP-K也算是过滤的一个种类。

在面试时，TOP-K算法也算是基础题目之一，分而治之的思想就能顺利解决。如果加上内存、网络IO、时间等限制的话，根据具体的情况，分治也是一个不错的方法。这里当然不是讨论这个算法的解决方法，而是讨论在MapReduce框架中怎么解决这个问题。

class Mapper
  method setup
    H = new Fixed Array(K)
  method map(Value val)
    H.add(val)
    if(len(H) > K)
      truncate H to K size
  method cleanup
    for all Value val in H do:
      emit(null, val)
    
class Reducer
  method setup
    H = new Fixed Array(K)
  method reduce(Value key, Value[val1,val2,...])
    sort [val1,val2,...]
    truncate val to top K to H
    for all Value val in H do
    emit(null,val)

 

需要注意的是，所有的Mapper最后都汇总到同一个Reducer上，也就是说必须限制Reducer数目为1。那么最后Reducer需要处理的数据位K*M，这个K*M的问题后面再说。

 

TOP-K算法很简单，在很多情况下都能运行的很好，这里不再赘述，下面我们讨论下这个算法的不足之处，不足之处很大程度上来自于单个Reducer：

 

1：多个Mapper输出结果到同一个Reducer中，如果K*M太大的话，Shuffle、Sort过程会比较耗时，这个是在本地硬盘上做的工作，而不是在内存中。

2：单个Reducer运行的数据来自于各个Mapper，这样会导致单个Reducer成为热点，网络IO过大。

3：将所有的K*M个记录状态装载到内存中，如果数据量过大的话，这个装载时间可能会特别长

4：在内存里面Sort可能会导致虚拟机崩溃；不过这个不是问题，这个K*M的数据规模跟原始数据比算是小的了，可以使用打擂台或者最大（小）堆来维护K个记录

5：使用单个Reducer，没有使用Hadoop的并行写的能力，不过这个也不是个问题

 

单个Reducer的限制确实很大，有可能会导致该程序效率低下、不发正常工作甚至崩溃。对于单个Mapper程序，M的数量是可以限制的，如果K的数目过大的话，就需要额外的考虑了。不过总体来说，这个M值即使上万都没有关系。换句话说，Top-10000是干什么用的呢？甚至说有时候我们只关心TOP-K(K < 100)的内容。

 

最后还有个Distinct内容。

Distinct，顾名思义，数据去重；含义和SQL中的distinct含义一致。除去去重之外，Distinct还有一个非常重要的功能：防止Inner Join数据爆炸；distinct在数据做Join模式时能够发挥巨大的作用。

Distinct的代码比较简单，但是其功能绝对不简单，尤其是在做Join操作时。代码如下：

class Mapper
  method map(key,val)
    emit(key,null)
    
class Reducer
  method reduce(key, records)
    emit(key,null)

这个Distinct的典型应用就是UV的计算，这个算是比较熟悉的内容了，不再多做解释。

  

Filtering模式到这里就算是结束了，主要是通过减小集合的内容，来更近、更关注的观察数据。熟悉、熟练掌握这种模式，能够加深对于数据集合的理解和应用。