一般的な MapReduce データマイニングアルゴリズム: 平均と分散

平均と分散のマップ削減

一連の数値の平均と分散の式は誰にとっても明らかだと思います。マップ関数とリデュース関数の設計方法については、計算式から始めることができます。n 個の数値、つまり a1、a2....an があるとすると、平均 m=(a1+a2+...an) / n、分散 s= [(a1-m)^2+(a2-m)^2+....+(an-m)^2] / n となります。

分散式を S=[(a1^2+.....an^2)+m^m*n-2*m*(a1+a2+....an) ] / n に展開します。これに基づいて、マップ エンドの入力を (key, a1) に、出力を (1, (n1, sum1, var1)) に設定できます。n1 は各ワーカーによって計算された数値の数を表し、sum1 はこれらの数値の合計 (たとえば、a1+a2+a3...)、var1 はこれらの数値の二乗の合計 (たとえば、a1^2+a2^2+...) です。

この情報を受け取った後、reduce 側は直ちにすべての入力 n1、n2... を加算して n を取得し、sum1、sum2... を加算して sum を取得し、平均 m=sum/n を計算し、var1、var2... を加算して var を取得し、分散 S=(var+m^2*n-2*m*sum)/n を計算し、reduce 出力は (1,(m,S)) になります。

アルゴリズムコードは、mrjob の実装に基づいています (https://pythonhosted.org/mrjob/、機械学習実践の第 15 章)

 mrjob.job から MRJob をインポートします
クラス MRmean(MRJob):
    def __init__(self, *args, **kwargs):
        super(MRmean, self).__init__(*args, **kwargs)
自己入力カウント= 0  
自己入力合計= 0  
自己.inSqSum = 0  
    def map(self, key, val): # ちょうど2つの引数が必要
        Falseの場合: 譲歩
inVal =浮動小数点数(val)
        自己入力カウント += 1
        self.inSum += inVal #各要素の合計
        self.inSqSum += inVal*inVal #各要素を2乗する
    def map_final(self):
 mn =自己.inSum / 自己.inCount
 mnSq =自己.inSqSum / 自己.inCount
        yield (1, [self.inCount, mn, mnSq]) #出力をマップしますが、ここではmn = sum1 /mn、 mnsq = var1 /mn です。
    def Reduce(自己、キー、パックされた値):
 cumVal = 0.0 ; cumSumSq = 0.0 ; cumN = 0.0
        for valArr in packedValues: #ストリーム入力から値を取得するマップ側の出力を解析する
nj =浮動小数点数(valArr[0])
            cumN += nj
            cumVal += nj*float(valArr[1])
            cumSumSq += nj*float(valArr[2])
平均= cumVal /cumN
 var = (cumSumSq - 2*平均*cumVal + cumN*平均*平均)/cumN
        yield (mean, var) #meanとvarの出力を出力します。
   defステップ(self):
        戻り値 ([self.mr( mapper = self .map, mapper_final = self .map_final,\
レデューサー= self .reduce,)])
 __name__ == '__main__' の場合:
    MRmean.run()