更新 15项目案例

src/python/15.BigData_MapReduce/mrSVM.py

@@ -1,3 +1,5 @@
+#!/usr/bin/python
+# coding:utf8
 '''
 Created on 2017-04-07
 MapReduce version of Pegasos SVM
@@ -9,19 +11,20 @@ from mrjob.job import MRJob
 import pickle
 from numpy import *
 
+
 class MRsvm(MRJob):
     DEFAULT_INPUT_PROTOCOL = 'json_value'
-    
+
     def __init__(self, *args, **kwargs):
         super(MRsvm, self).__init__(*args, **kwargs)
-        self.data = pickle.load(open('C:\Users\Peter\machinelearninginaction\Ch15\svmDat27'))
+        self.data = pickle.load(open('input/15.BigData_MapReduce/svmDat27'))
         self.w = 0
         self.eta = 0.69
         self.dataList = []
         self.k = self.options.batchsize
         self.numMappers = 1
-        self.t = 1  #iteration number
-                                                 
+        self.t = 1  # iteration number
+                                      
     def configure_options(self):
         super(MRsvm, self).configure_options()
         self.add_passthrough_option(
@@ -30,49 +33,61 @@ class MRsvm(MRJob):
         self.add_passthrough_option(
             '--batchsize', dest='batchsize', default=100, type='int',
             help='k: number of data points in a batch')
-    
-    def map(self, mapperId, inVals): # 需要 2 个参数
-        #input: nodeId, ('w', w-vector) OR nodeId, ('x', int)
-        if False: yield
-        if inVals[0]=='w':                  # 积累 w向量
+
+    def map(self, mapperId, inVals):  # 需要 2 个参数
+        # input: nodeId, ('w', w-vector) OR nodeId, ('x', int)
+        if False:
+            yield
+        if inVals[0] == 'w':                  # 积累 w向量
             self.w = inVals[1]
-        elif inVals[0]=='x':
-            self.dataList.append(inVals[1])# 累积数据点计算
-        elif inVals[0]=='t': self.t = inVals[1] 
-        else: self.eta=inVals # 这用于 debug， eta未在map中使用
-        
+        elif inVals[0] == 'x':
+            self.dataList.append(inVals[1])   # 累积数据点计算
+        elif inVals[0] == 't':
+            self.t = inVals[1]
+        else:
+            self.eta = inVals  # 这用于 debug， eta未在map中使用
+
     def map_fin(self):
-        labels = self.data[:,-1]; X=self.data[:,0:-1]# 将数据重新形成 X 和 Y
-        if self.w == 0: self.w = [0.001]*shape(X)[1] # 在第一次迭代时，初始化 w
+        labels = self.data[:,-1]
+        X = self.data[:, 0:-1]   # 将数据重新形成 X 和 Y
+        if self.w == 0: 
+            self.w = [0.001] * shape(X)[1]   # 在第一次迭代时，初始化 w
         for index in self.dataList:
-            p = mat(self.w)*X[index,:].T #calc p=w*dataSet[key].T 
+            p = mat(self.w)*X[index, :].T    # calc p=w*dataSet[key].T 
             if labels[index]*p < 1.0:
-                yield (1, ['u', index])# 确保一切数据包含相同的key                           
-        yield (1, ['w', self.w])       # 它们将在同一个 reducer
+                yield (1, ['u', index])      # 确保一切数据包含相同的key                           
+        yield (1, ['w', self.w])             # 它们将在同一个 reducer
         yield (1, ['t', self.t])
 
     def reduce(self, _, packedVals):
-        for valArr in packedVals: # 从流输入获取值
-            if valArr[0]=='u':  self.dataList.append(valArr[1])
-            elif valArr[0]=='w': self.w = valArr[1]
-            elif valArr[0]=='t':  self.t = valArr[1] 
-        labels = self.data[:,-1]; X=self.data[:,0:-1]
-        wMat = mat(self.w);   wDelta = mat(zeros(len(self.w)))
+        for valArr in packedVals:            # 从流输入获取值
+            if valArr[0] == 'u':
+                self.dataList.append(valArr[1])
+            elif valArr[0] == 'w':
+                self.w = valArr[1]
+            elif valArr[0] == 't':
+                self.t = valArr[1] 
+
+        labels = self.data[:, -1]
+        X = self.data[:, 0:-1]
+        wMat = mat(self.w)
+        wDelta = mat(zeros(len(self.w)))
+
         for index in self.dataList:
-            wDelta += float(labels[index])*X[index,:] #wDelta += label*dataSet
-        eta = 1.0/(2.0*self.t)       #calc new: eta
-        #calc new: w = (1.0 - 1/t)*w + (eta/k)*wDelta
+            wDelta += float(labels[index]) * X[index, :]  # wDelta += label*dataSet
+        eta = 1.0/(2.0*self.t)       # calc new: eta
+        # calc new: w = (1.0 - 1/t)*w + (eta/k)*wDelta
         wMat = (1.0 - 1.0/self.t)*wMat + (eta/self.k)*wDelta
-        for mapperNum in range(1,self.numMappers+1):
-            yield (mapperNum, ['w', wMat.tolist()[0] ]) #发出 w
+        for mapperNum in range(1, self.numMappers+1):
+            yield (mapperNum, ['w', wMat.tolist()[0]])    # 发出 w
             if self.t < self.options.iterations:
-                yield (mapperNum, ['t', self.t+1])# 增量 T
-                for j in range(self.k/self.numMappers):#emit random ints for mappers iid
-                    yield (mapperNum, ['x', random.randint(shape(self.data)[0]) ])
-        
+                yield (mapperNum, ['t', self.t+1])        # 增量 T
+                for j in range(self.k/self.numMappers):   # emit random ints for mappers iid
+                    yield (mapperNum, ['x', random.randint(shape(self.data)[0])])
+
     def steps(self):
-        return ([self.mr(mapper=self.map, reducer=self.reduce, 
-                         mapper_final=self.map_fin)]*self.options.iterations)
+        return ([self.mr(mapper=self.map, reducer=self.reduce, mapper_final=self.map_fin)] * self.options.iterations)
+
 
 if __name__ == '__main__':
     MRsvm.run()