开发笔记:用hadoop实现SimRank++算法权值转移矩阵的计算

本文主要针对广告检索领域的查询重写应用，依据查询-广告点击二部图，在MapReduce框架上实现SimRank++算法。关于SimRank++算法的背景和原理请參看前一篇文章《基于MapReduce的SimRank++算法研究与实现》。

SimRank++的矩阵形式的计算公式为：

算法主要过程例如以下：
Step1: 计算权值矩阵。并获取最大Query编号和最大广告编号。
Step2: 以Step1的输出作为输入，迭代计算SimRank相似度。
Step3: 计算证据矩阵。并用计算结果修正Step2的输出，计算出终于的经过归一化的相似度分数。
Step4: 把Step3的输出转化为固定的格式，得到终于的相似度分数结果。
当中Step2迭代计算SimRank相似度分数比較复杂。由一系列的MapReduce作业组成。

本文主要关注Step1。即计算权值矩阵的计算。Step2~4将在兴许的文章中给出。

1.输入文件的格式

为了简单起见，在我们的实现中。用点击次数作为边的权值。

一个更好的选择是用广告点击率(Click Through Rate, CTR)作为权值。理由例如以下：某个广告在q1下展示10000次。点击100次（CTR为0.01）。在q2下展示1000次，点击90次（CTR为0.09）；在q3下展示1000次。点击100次（CTR为0.1）；显然q3和q2的相似度要比q3和q1的相似度要高，然而假设仅仅考虑点击次数，那么算法会觉得q3和q1的相似度比q3和q2的高。

期待的输入数据的文件格式：
1. Query和广告的点击关系数据文件（下面记为qas文件）的每一行的格式例如以下：
qas ^A queryid { ^A adid ^B clicknum}
当中。{ }表示内部的内容能够反复1次或多次，但至少一次；“qas”的标识字符串；‘^A’是ASCII码为1的不可见字符。‘^B’是ASCII码为2的不可见字符。
2. 广告和Query的点击关系数据文件（下面记为aqs文件）的每一行的格式例如以下：
aqs ^A adid { ^A queryid ^B clicknum}
当中，{ }表示内部的内容能够反复1次或多次，但至少一次。“aqs”的标识字符串；‘^A’是ASCII码为1的不可见字符；‘^B’是ASCII码为2的不可见字符。

上图所看到的的查询和广告之间的点击关系相应的文件格式例如以下：

qas文件
qas ^A 1 ^A 1 ^B 10 ^A 3 ^B 5
qas ^A 2 ^A 2 ^B 7 ^A 3 ^B 6
aqs文件
aqs ^A 1 ^A 1 ^B 10
aqs ^A 2 ^A 2 ^B 7
aqs ^A 3 ^A 1 ^B 5 ^A 2 ^B 6


2. 思路分析

权值矩阵元素的计算公式为：

能够看出。 variance(a)的计算须要用到aqs文件， normalize_weight(q,a)的计算须要用到qas文件； variance(q)的计算须要用到qas文件， normalize(q,a)的计算须要用到aqs文件。从而，在计算W(a,q) 和 W(q,a)时都要用到aqs文件和qas文件。这使得MapReduce算法的设计比較困难。

考虑前面所述的一个简单样例。”Mapper”任务在处理qas文件时会计算出例如以下所看到的的内容。

”Mapper”任务在处理aqs文件时会计算出例如以下所看到的的内容。

在计算W(q,a)时须要使用到variance(a)和normalizedweight(q, a)；在计算W(a,q)时须要使用到variance(q)和normalizedweight(a, q)。

因此，依据以上分析，对于一个特定的q和a。须要把Map任务的输出中的variance(a)和normalizedweight(q, a)”Shuffle”到同一个”Reduce”节点，由该”Reduce”节点计算出W(q,a)。同理，须要把”Map”任务的输出中的variance(q)和normalizedweight(a,
q) ”Shuffle”到同一个”Reduce”节点。由该”Reduce”节点计算出W(a,q)。

另外。能够看出。在计算W(q1,a), W(q2,a),……. 时都须要用到variance(a)，因此我们希望计算 的“Reduce”节点接受到的值列表中variance(a)项排在全部normalized_weight(q, a)项之前。

MapReduce框架在记录到达”Reducer”之前按键对记录排序，但键所相应的值并没有被排序。因为值来自不同的map任务，所以在多次执行程序时，值的出现顺序并不固定。导致每次执行作业的时间会各有不同。一般来说，大多数MapReduce程序无需考虑值在”Reduce”函数中出现的顺序。可是。像我们这里碰到的情况一样。有时确实须要通过对键进行排序和分组等以实现对值的排序。通过MapReduce框架辅助对记录值排序的方法总结例如以下：
(1) 定义包含自然键和自然值的组合键。

(2) 键的comparator依据组合键对记录进行排序，即同一时候利用自然键和自然值进行排序。
(3) 针对组合键partitioner和分组comparator在进行分区和分组时均仅仅考虑自然键。

基于以上分析。计算权值矩阵的MapReduce任务须要小心地设计”Map”任务输出的Key和Value的数据结构以及”Partitioner”函数。

3. 算法实现

(1) Map任务输出的键(Key)的数据结构
键(Key)由一个三元组构成：
type用于标识index1是广告的编号(0)，还是Query的编号(1)；当type = 0时。相应的值(value)表示normalizedweight(q,a)，当中q等于index1，a等于index2；当type = 1时。value表示normalizedweight(a,q)，当中a等于index1，q等于index2；
另外，当index2 = -1时。表示相应的值为方差（variance(index1)）。设为-1是为了保证同一组Key相应的值列表中方差项排在第一个。
键(Key)的三个元素都參与comparator的比較。

(2) Map任务输出的值(Value)的数据结构
值(Value)有一个二元组构成：。当中index总是等于相应的键的第三个元素index2。这里看似冗余，事实上不然。

由于我们在利用MapReduce框架辅助排序时，分组函数(GroupComparator)仅仅比較Key的前两个元素，忽略Key的第三个元素，这样仅仅有Key的前两个元素的值同样，那么他们的值将合并到同一个列表中。有唯一的一个“Reduce”函数处理。MapReduce框架在默认情况下仅仅会把key全然同样值合并到同一个列表中。

因此我们须要设置OutputValueGroupingComparator为我们自己定义的GroupComparator。能够利用例如以下的语句设置：

conf.setOutputValueGroupingComparator(GroupComparator.class);


(3) 分区函数
分区函数控制”Map”任务的每一个输出记录由哪一个”Reduce”节点处理。

在权值矩阵的计算作业中该函数的地位特别重要。

依据上一小节的分析和辅助排序的要求，分区函数仅仅考虑键的前两个元素。

我们把”Reduce”节点分成两部分，一部分计算 ，另外一部分计算 。”Partition”函数的代码例如以下。

public int getPartition(Key key, Value value, int numPartitions)
{
int offset = numPartitions / 2;
if (key.type == 0)
{
int base = numPartitions - offset;
return key.index1 % base + offset;
}
return key.index1 % offset;
}


(4) “Map”函数和”Reduce”函数
“Map”函数和”Reduce”函数并行地处理基本的工作。当中”Map”函数读入qas文件，计算出variance(q)和normalizedweight(a, q)。读入aqs文件。输出variance(a)和normalizedweight(q, a)。

同一时候为了以后的计算方便，”Map”函数还记录下最大的Query编号和最大的Ad编号。

因为多个”Map”函数之间不能相互通信。为了得到全局的最大Query编号和Ad编号。每一个Map函数结束的时候在一个指定的HDFS文件夹下新建一个以本函数统计出的当前最大编号为文件名称的空文件，前提条件是此时该指定文件夹下还没有更大编号的文件存在。

“Reduce”函数比較简单，直接依据公式计算出终于的权值就能够了。

”Reduce”输出的Key是一个二元组。表示权值矩阵的行号和列号。输出的值为对应的权值。

因为我们在同一个作业中同一时候计算了Query-Ad的权值矩阵和Ad-Query的权值矩阵。这两个矩阵在后面的SimRank实现过程中需要单独使用，因此必需要把两种的输出区分开来。我们使用MultipleOutputs类定义了两个数据收集对象，这两个数据收集对象输出的文件名称有不同的前缀。

“Mapper”和”Reducer”的伪代码例如以下。

计算权值矩阵的”Map”函数

Setup(){
currMaxQueryId ← 0
currMaxAdId ← 0
dir ← “hdfs://namenode/…/XXX”
}
Map(line_no, line_txt){
content ← Parser(line_txt)
if (content.type == 1)
currMaxQueryId ← max(currMaxQueryId, content. id)
else
currMaxAdId ← max(currMaxAdId, content. id)
weight_sum ← sum(content.weights)
variance ← var(content.weights)
emit , <-1, variance>
for e in content.elements
normalized_weight ← e.weight / seight_sum
emit ,
}
Close(){
Query_id ← getCurrentQueryId(dir)
Ad_id ← getCurrentAdId(dir)
If (currMaxQueryId > Query_id)
Touch dir/ currMaxQueryId
If (currMaxAdId > Ad_id)
Touch dir/ currMaxAdId
}


计算权值矩阵的”Reduce”函数

Reduce(key, valueList){
variance ← valueList[0]
spread ← exp(-variance)
for v in valueList[1]…valueList[N]
emit , spread * v.value
}