Hadoop分布式环境下的数据抽样

xiaochuan

　　1. 问题由来
　　Google曾经有一道非常经典的面试题：

　　给你一个长度为N的链表。N很大，但你不知道N有多大。你的任务是从这N个元素中随机取出k个元素。你只能遍历这个链表一次。你的算法必须保证取出的元素恰好有k个，且它们是完全随机的（出现概率均等）？

　　这道题的解法非常多，网上讨论也非常热烈。本文要讨论的是，这个问题是从何而来，有什么实用价值？
　　自从有了Hadoop之后，该问题便有了新的应用载体。随着数据量的增多，很多数据挖掘算法被转移到MapReduce上实现，而数据挖掘中有个基本的问题是怎样对数据进行抽样。在Hadoop中，每个job会被分解成多个task并行计算，而数据的总量事先是不知道的（知道job运行结束才能获取数总数，而数据量非常大时，扫描一遍数据的代价非常高），用户知道的只是要获取的样本量，那怎样在类似于Hadoop的分布式平台上进行数据抽样？
　　回过头来看google的这道面试题，是不是正好时Hadoop平台上海量数据抽样问题？
　　2. 在Hadoop上编写抽样程序
　　2.1 解法一
　　(1) 设计思想
　　蓄水池抽样：先保存前k个元素， 从第k+1个元素开始， 以k/i (i=k+1, k+2,…,N) 的概率选中第i个元素，并随机替换掉一个已保存的记录，这样遍历一次得到k个元素，可以保证完全随机选取。
　　(2) MapReduce实现
　　要实现该抽样算法，只需编写Mapper即可。在Map函数中，用户定义一个vector保存选中的k个元素，待扫描完所有元素后，在析构函数中将vector中的数据写到磁盘中。
　　用户运行job时，需指定每个map task的采样量。比如，该job的map task个数为s，则每个map task需要采集k/s个元素。
　　(3) 优缺点分析
　　由于该job没有reduce task，因而效率很高。然而，该方法选中某个元素的概率并不完全是k/N!
　　2.2 解法二
　　(1) 设计思想
　　依次扫描每个元素，为每个元素赋予一个随机的整数值；然后使用Top K算法（譬如最大K个整数）得到需要的K个元素。
　　(2) MapReduce实现
　　要实现该算法，用户需要编写mapper和reducer，在map函数中，为每个元素赋予一个随机数，将该随机数作为key，并将key最大的前k个保存下来（可使用小顶堆）；在reduce函数中，唯一的reduce task输出前k个元素。
　　(3) 优缺点分析
　　该算法比第一种算法低效，但由于整个过程自然流畅，实现起来非常简单，不易出错。
　　2.3 解法三
　　(1) 设计思想
　　考虑第一个元素，其以K/N的概率被选中；如果该节点被选中，则从剩下的(N-1)个元素中选出(K-1)个元素；如果没有被选中，则从剩下的(N-1)个元素中选出K个元素,…,依次这样下去，直到获取K个元素。
　　(2) MapReduce实现
　　用户只需编写Mapper即可。首先要获取每个map task输入的数据量，这个可以在InputFormat中计算得到。然后，在每个map函数中，采集k/s（其中s为map task数据量）个元素。
　　(3) 优缺点分析
　　由于该算法没有reduce task，效率比较高，但需要在InputFormat中统计数据量，编程复杂度较高。
　　3. 延伸
　　这个问题与《编程珠玑》上讨论的问题很相似：
　　输入两个整数m和n，其中m<n。输出是0~n-1范围内m个随机整数的有序表，不允许重复。
　　对于该问题，大致存在四种算法，他们有不同的优缺点。
　　(1) 第一种方法来自Knuth的《The art of Computer Programming, Volume 2: Seminumerical Algorithms》
　　伪代码是：

1 2 3 4 5 6 7 8 9 10 11 12 13 14

select = m remaining = n for I = [0 n ) if(bigrand() % remaining) < select print i select— remaining—

　　只要m<=n，程序选出来的整数就恰为m个。
　　C++的实现如下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

void genKnuth(int m, int n) { for(inti = 0; i < n; i++) { if(bigrand() % (n - i) < m） { cout <<i << endl; m--; } } }

　　该算法非常节省空间，但需要全部扫描n个数，当n很多时，效率不高。
　　（2）第二种方法的复杂度只与m有关，采用了set（实际上是红黑树）节省时间。代码如下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14

void gensets(int m,intn) { set<int> S; while(S.size() < m) { S.insert(bigrand() % n); } // print S }

　　该方法每次插入均在O(log m)时间内完成，但需要的空间开销很大。
　　（3）第三种方法克服了（2）的缺点，代码如下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26

void genshuf(int m,intn) { inti, j; int*x =newint[n]; for(i = 0; i < n; i++) { x = i; } for(i = 0; i < m; i++) { j = randint(i, n-1); intt = x; x = x[j]; x[j] = x; } sort(x, x+m); //print result }

　　该算法需要n个元素的内存空间和O(n+mlogm)的时间，其性能通常不吐Knuth的算法。
　　（4）当m接近n时，基于集合的算法生成的很多随机数都要丢掉，因为之前的数已经存在于集合中了，为了改进这一点，算法如下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28

void genfloyd(int m, int n) { set<int> S; set<int>::iterator i; for(intj=n-m; j < n; j++) { intt = bigrand()%（j+1); if(S.find(t) == S.end()){ S.insert(t);// t not in S }else{ S.insert(j);// t in S } } //print results }

　　4. 参考资料
　　(1) 《编程珠玑》第二版
　　(2) http://hi.baidu.com/sunxiangwei/blog/item/a7839b51bdbf522e42a75b45.html
　　本文链接地址: http://dongxicheng.org/data-mining/hadoop-sampling/