multiprocessing で MapReduce を実装する

Pool クラスは1サーバで簡単な MapReduce 実装を作成するために使用することができます。それは分散処理の最大限の利点を得るものではないですが、ある問題を個々の分散処理の単位に分解することがどのぐらい簡単かを分かり易く説明します。

SimpleMapReduce

MapReduce ベースのシステムでは、入力データを複数のワーカーインスタンスで処理するためにチャンクに分解します。各入力データのチャンクは簡単な変換を行って中間状態に map されます。中間データはまとめて集約されて、全ての関連する値が一緒になるように key value に基づいて分割されます。最後に、分割されたデータは結果セットへ reduce されます。

import collections
import itertools
import multiprocessing

class SimpleMapReduce(object):
    
    def __init__(self, map_func, reduce_func, num_workers=None):
        """
        map_func

          Function to map inputs to intermediate data. Takes as
          argument one input value and returns a tuple with the key
          and a value to be reduced.
        
        reduce_func

          Function to reduce partitioned version of intermediate data
          to final output. Takes as argument a key as produced by
          map_func and a sequence of the values associated with that
          key.
         
        num_workers

          The number of workers to create in the pool. Defaults to the
          number of CPUs available on the current host.
        """
        self.map_func = map_func
        self.reduce_func = reduce_func
        self.pool = multiprocessing.Pool(num_workers)
    
    def partition(self, mapped_values):
        """Organize the mapped values by their key.
        Returns an unsorted sequence of tuples with a key and a sequence of values.
        """
        partitioned_data = collections.defaultdict(list)
        for key, value in mapped_values:
            partitioned_data[key].append(value)
        return partitioned_data.items()
    
    def __call__(self, inputs, chunksize=1):
        """Process the inputs through the map and reduce functions given.
        
        inputs
          An iterable containing the input data to be processed.
        
        chunksize=1
          The portion of the input data to hand to each worker.  This
          can be used to tune performance during the mapping phase.
        """
        map_responses = self.pool.map(self.map_func, inputs, chunksize=chunksize)
        partitioned_data = self.partition(itertools.chain(*map_responses))
        reduced_values = self.pool.map(self.reduce_func, partitioned_data)
        return reduced_values

ファイルの単語を数える

次のサンプルスクリプトは、マークアップの幾つかを無視して本稿の reStructuredText のソースファイル内の “単語” を数える SimpleMapReduce を使用します。

import multiprocessing
import string

from multiprocessing_mapreduce import SimpleMapReduce

def file_to_words(filename):
    """Read a file and return a sequence of (word, occurances) values.
    """
    STOP_WORDS = set([
            'a', 'an', 'and', 'are', 'as', 'be', 'by', 'for', 'if', 'in', 
            'is', 'it', 'of', 'or', 'py', 'rst', 'that', 'the', 'to', 'with',
            ])
    TR = string.maketrans(string.punctuation, ' ' * len(string.punctuation))

    print multiprocessing.current_process().name, 'reading', filename
    output = []

    with open(filename, 'rt') as f:
        for line in f:
            if line.lstrip().startswith('..'): # rst コメント行を読み飛ばす
                continue
            line = line.translate(TR) # 句読点を取り除く
            for word in line.split():
                word = word.lower()
                if word.isalpha() and word not in STOP_WORDS:
                    output.append( (word, 1) )
    return output


def count_words(item):
    """Convert the partitioned data for a word to a
    tuple containing the word and the number of occurances.
    """
    word, occurances = item
    return (word, sum(occurances))


if __name__ == '__main__':
    import operator
    import glob

    input_files = glob.glob('*.rst')
    
    mapper = SimpleMapReduce(file_to_words, count_words)
    word_counts = mapper(input_files)
    word_counts.sort(key=operator.itemgetter(1))
    word_counts.reverse()
    
    print '\nTOP 20 WORDS BY FREQUENCY\n'
    top20 = word_counts[:20]
    longest = max(len(word) for word, count in top20)
    for word, count in top20:
        print '%-*s: %5s' % (longest+1, word, count)

file_to_words() はそれぞれの入力ファイルを単語と数値1(1つ出現したことを表す)を持つタプルのシーケンスに変換します。そのデータはキーのようにその単語を使用して partition() で分割されます。そのため、分割したデータはその単語の出現頻度を表す1つの値のシーケンスとキーで構成されます。reduce フェーズでは、分割したデータは count_words() により単語とその単語の出現数を含むタプルのセットに変換されます。

$ python multiprocessing_wordcount.py
PoolWorker-2 reading communication.rst
PoolWorker-3 reading index.rst
PoolWorker-1 reading basics.rst
PoolWorker-4 reading mapreduce.rst

TOP 20 WORDS BY FREQUENCY

process         :    84
multiprocessing :    53
starting        :    52
class           :    49
func            :    39
worker          :    39
mod             :    33
poolworker      :    32
after           :    32
running         :    31
consumer        :    31
exiting         :    30
processes       :    30
start           :    29
python          :    28
literal         :    26
header          :    26
pymotw          :    26
end             :    26
daemon          :    23

See also

MapReduce - Wikipedia
Wikipedia の MapReduce 概要
MapReduce: Simplified Data Processing on Large Clusters
MapReduce に関する Google Labs のプレゼンテーションと論文
operator
itemgetter() のようなオペレータツール

本稿のレビューを手伝ってくれた Jesse Noller に感謝します。

Bookmark and Share