程序清單3 分布式Pegasos算法的mapper和reducer代碼

程序清單3 里的第一個(gè)方法是map()，這也是分布式的部分，它得到輸入值并存儲(chǔ)，以便在map_fin()中處理。該方法支持三種類型的輸入：w向量、t或者x。t是迭代次數(shù)，在本方法中不參與運(yùn)算。狀態(tài)不能保存，因此如果需要在每次迭代時(shí)保存任何變量并留給下一次迭代，可以使用key/value對(duì)傳遞該值，抑或是將其保存在磁盤(pán)上。顯然前者更容易實(shí)現(xiàn)，速度也更快。

map_fin()方法在所有輸入到達(dá)后開(kāi)始執(zhí)行。這時(shí)已經(jīng)獲得了權(quán)重向量w和本次批處理中的一組x值。每個(gè)x值是一個(gè)整數(shù)，它并不是數(shù)據(jù)本身，而是索引。數(shù)據(jù)存儲(chǔ)在磁盤(pán)上，當(dāng)腳本執(zhí)行的時(shí)候讀入到內(nèi)存中。當(dāng)map_fin()啟動(dòng)時(shí)，它首先將數(shù)據(jù)分成標(biāo)簽和數(shù)據(jù)，然后在本次批處理的數(shù)據(jù)（存儲(chǔ)在self.dataList里）上進(jìn)行迭代，如果有任何值被錯(cuò)分就將其輸出給reducer。為了在mapper和reducer之間保存狀態(tài)，w向量和t值都應(yīng)被發(fā)送給reducer。

最后是reduce()函數(shù)，對(duì)應(yīng)本例只有一個(gè)reducer執(zhí)行。該函數(shù)首先迭代所有的key/value對(duì)并將值解包到一個(gè)局部變量datalist里。之后dataList里的值都將用于更新權(quán)重向量w，更新量在wDelta中完成累加。然后，wMat按照wDelta和學(xué)習(xí)率eta進(jìn)行更新。在wMat更新完畢后，又可以重新開(kāi)始整個(gè)過(guò)程：一個(gè)新的批處理過(guò)程開(kāi)始，隨機(jī)選擇一組向量并輸出。注意，這些向量的key是mapper編號(hào)。

為了看一下該算法的執(zhí)行效果，還需要用一些類似于reducer輸出的數(shù)據(jù)作為輸入數(shù)據(jù)啟動(dòng)該任務(wù)，我為此附上了一個(gè)文件kickStart.txt。在本機(jī)上執(zhí)行前面的代碼可以用下面的命令：

%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

這樣就輸出了結(jié)果。經(jīng)過(guò)2次和50次迭代后的分類面如圖1所示。

圖1 經(jīng)過(guò)多次迭代的分布式Pegasos算法執(zhí)行結(jié)果。該算法收斂迅速，多次迭代后可以得到更好的結(jié)果

如果想在EMR上運(yùn)行該任務(wù)，可以添加運(yùn)行參數(shù)：-r emr。該作業(yè)默認(rèn)使用的服務(wù)器個(gè)數(shù)是1。如果要調(diào)整的話，添加運(yùn)行參數(shù)：--num-ec2-instances=2(這里的2也可以是其他正整數(shù)），整個(gè)命令如下：

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

要查看所有可用的運(yùn)行參數(shù)，輸入%python mrsvM.py -h。

責(zé)任編輯：admin