MapReduce 基础算法程序设计(3)：单词共现算法

问题导读

1.什么是单词共现算法？
2.单词共现算法如何实现?
3.单词共现算法实现需要注意哪些问题？

1、单词共现算法的基本设计

单词共现算法是MapReduce可以用来高效解决的一大类问题的抽象化描述。在自然语言处理以及简历语料库上也有着重要的应用。其目的是在海量语料库中发现在固定窗口内单词a和单词b共同出现的频率，从而构建单词共现矩阵，这样的矩阵可以是对称的，也可以是不对称的，这要看具体的应用。

                   这种抽象化的任务的有效解决在实际生活中有着很多的应用。例如电子商家希望发现不同物品被同时购买的情况以便有效安排货物的摆放位置；同时对信息检索领域同义词词典的构建以及文本挖掘等都有着重要的实际应用价值。

                   设有一个英文句：

                   weare not what we want to be but at least we are not what we used to be.

                   设共现窗口定义为连续出现的两个单词，则表中 给出了上局英文的共现矩阵。

示例英文语句的共现矩阵

	we	are	not	what	we	want	to	be	but	at	least	used
we		2				1						1
are	2		2
not		2		2
what			2
want	1						1
to						1		1				1
be							1
but										1
at									1
least
used	1						1

2、单词共现算法的实现
这里我们利用了MapReduce实现了单词共现算法中的一种称作pairs的算法。其算法伪代码如下：
Map端伪代码如下：

上述Mapper伪代码中使用了一个Window定义，表示如果单词u属于单词w的窗口内，则认为是(u, w)的一次共现。这里窗口Window可以根据不同的应用需求有不同的定义，比例，可定义为一个固定大小的窗口，或者是前后相连出现，在同一句中出现、在统一段落中出现的单词等。
        例如，如果窗口中的单词为[w1, w2, w3]，我们发射((w1, w2),1)和((w1, w3),1)出去然后窗口向后移动一个单词，Reduce阶段则对发来的相同键的值进行简单的求和即可。这里单词顺序有无关系需要看具体的情况而定。另外，在实际实现中我们需要传入Map的数据是以一个文本为单位的，这里需要实现一个WholeFileInputFormat以便一个文本不被拆分被整个传入到一个Map节点。这里我们以一个具体的例子来说明算法的工作流程，例如一个文档中的内容，如下图所示：





Map阶段窗口在文本上滑动过程

例如在Maori阶段，一个Map节点接收到如图所示的一个文档的全部内容，窗口大小为7，那么首先窗口先覆盖了Mapreduce is a new technique to process big data，然后该节点将键值对((Mapreduce, is),1)(( Mapreduce, a),1),(( Mapreduce, new),1),(( Mapreduce, technique),1),(( Mapreduce, to),1),(( Mapreduce, process),1)发射出去，随后窗口向后滑动一格，与上面相似，这时将((is, a),1),((is, new),1),((is technique),1),((is, to),1),((is, process),1),((is, big),1)射出出去。最后再向后滑动一格单词至文档的末尾，与上面相似，发送相应的键值对出去。当窗口尾部已经到达文档尾部时，滑动窗口则通过将窗口头部向后”缩进”来进行，此过程一直进行到窗口大小为2停止。

3、单词共现算法实现中的细节问题
        这里在具体时我们需要注意一些细节。首先我们发射出去的键是单词对而不再是一些基本数据类型，因此首先我们需要自定义一个WordPair类，该类需要实现WirtableComparable接口。另外，由于Hadoop默认使用的Partitioner是HashPartitioner，其计算方法为Reducer=(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks,以此得到所选择的Reduce节点。所以我们需要重写自定义类WordPair的hashCode()方法，使得相同的WordPair主键(不考虑顺序)都被发送到相同的Reduce节点去。另外，我们还需要重写compareTo()和equals()方法使得相同的WordPair键的值可以比较大小和排序。这里我们定义了新的类WordPair，其主要代码如下：

package Algorithm;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

public class WordPair implements WritableComparable<WordPair> {

        private String wordA;
        private String wordB;
        
        @Override
        public void write(DataOutput out) throws IOException {
                // TODO Auto-generated method stub
                
        }

        @Override
        public void readFields(DataInput in) throws IOException {
                // TODO Auto-generated method stub
                
        }

        @Override
        public int compareTo(WordPair wordPair) {
                // TODO Auto-generated method stub
                return 0;
        }
        @Override
        public int hashCode() {
                return (wordA.hashCode() + wordB.hashCode()) * 17;
        }
        @Override
        public boolean equals(Object o) {
                // 无序对,不用考虑顺序
                if(!(o instanceof WordPair)){
                        return false;
                }
                WordPair w = (WordPair)o;
                if((this.wordA.equals(w.wordA) && this.wordB.equals(w.wordB)) || (this.wordB.equals(w.wordA) && this.wordA.equals(w.wordB))){
                        return true;
                }
                return false;
        }

}
复制代码

而compareTo()方法则调用equals()方法来判断两个对象是否相等，至此我们重写了一些方法使得所有兼职对可以被正确地送到目的地Reduce节点。
        单词共现算法中Map端实现代码如下：

public void map(Text docName, BytesWritable docContent, Context context) throws IOException, InterruptedException {
                Matcher matcher = wordPattern.matcher(new String(docContent.getBytes(),"UTF-8"));
                
                while(matcher.find()){
                        windowQueue.add(matcher.group());
                        if(windowQueue.size() >= windowSize){
                                Iterator<String> it = windowQueue.iterator();
                                String w1 = it.next();
                                while(it.hasNext()){
                                        context.write(new WordPair(w1, next), one);
                                }
                                windowQueue.remove();
                        }
                }
                
                while(!(windowQueue.size() <=1)){
                        Iterator<String> it = windowQueue.iterator();
                        String w1 = it.next();
                        while(it.hasNext()){
                                context.write(new WordPair(w1, it.next()), one);
                        }
                        windowQueue.remove();
                }

        }
复制代码

而Reduce端的代码则比较简单，只是将同一个键的所有值进行简单的累加而已。这里不在给出具体代码，读者可以自己实现或参考代码中的实现。

相关帖子：
MapReduce 基础算法程序设计(1)

MapReduce 基础算法程序设计(2)

MapReduce 基础算法程序设计(3)：单词共现算法

谢谢楼主分享。