程序清单3 分布式Pegasos算法的mapper和reducer代码

程序清单3 里的第一个方法是map()，这也是分布式的部分，它得到输入值并存储，以便在map_fin()中处理。该方法支持三种类型的输入：w向量、t或者x。t是迭代次数，在本方法中不参与运算。状态不能保存，因此如果需要在每次迭代时保存任何变量并留给下一次迭代，可以使用key/value对传递该值，抑或是将其保存在磁盘上。显然前者更容易实现，速度也更快。

map_fin()方法在所有输入到达后开始执行。这时已经获得了权重向量w和本次批处理中的一组x值。每个x值是一个整数，它并不是数据本身，而是索引。数据存储在磁盘上，当脚本执行的时候读入到内存中。当map_fin()启动时，它首先将数据分成标签和数据，然后在本次批处理的数据（存储在self.dataList里）上进行迭代，如果有任何值被错分就将其输出给reducer。为了在mapper和reducer之间保存状态，w向量和t值都应被发送给reducer。

最后是reduce()函数，对应本例只有一个reducer执行。该函数首先迭代所有的key/value对并将值解包到一个局部变量datalist里。之后dataList里的值都将用于更新权重向量w，更新量在wDelta中完成累加。然后，wMat按照wDelta和学习率eta进行更新。在wMat更新完毕后，又可以重新开始整个过程：一个新的批处理过程开始，随机选择一组向量并输出。注意，这些向量的key是mapper编号。

为了看一下该算法的执行效果，还需要用一些类似于reducer输出的数据作为输入数据启动该任务，我为此附上了一个文件kickStart.txt。在本机上执行前面的代码可以用下面的命令：

%python mrSVM.py < kickStart.txt

streaming final output from c: \users\peter\appdata\local\temp

\mrSVM. Peter.20110301.011916. 373000 \output \part-00000

1             ["w*,[0.51349820499999987, -0.084934502500000009]]

removing tmp directory c: \users\peter\appdata\local\temp

\mrSVM.Peter.20110301.011916.373000

这样就输出了结果。经过2次和50次迭代后的分类面如图1所示。

图1 经过多次迭代的分布式Pegasos算法执行结果。该算法收敛迅速，多次迭代后可以得到更好的结果

如果想在EMR上运行该任务，可以添加运行参数：-r emr。该作业默认使用的服务器个数是1。如果要调整的话，添加运行参数：--num-ec2-instances=2(这里的2也可以是其他正整数），整个命令如下：

%python mrSVM.py -r emr --num-ec2-instances=3 < kickStart.txt > myLog.txt

要查看所有可用的运行参数，输入%python mrsvM.py -h。

责任编辑：admin