Word Frequency With Mapreduce in Python

“MapReduce is a programming framework popularized by Google and used to simplify data processing across massive data sets. As people rapidly increase their online activity and digital footprint, organizations are finding it vital to quickly analyze the huge amounts of data their customers and audiences generate to better understand and serve them.”

The algorithm fundamentally boils down to two crisp steps: map and reduce:

• map (k1, v1) -> list(k2, v2)
• reduce (k2, list(v2)) -> list(v2)

The canonical example of MapReduce is as below:

void map(String name, String document):
  // name: document name
  // document: document contents
  for each word w in document:
    EmitIntermediate(w, "1");

void reduce(String word, Iterator partialCounts):
  // word: a word
  // partialCounts: a list of aggregated partial counts
  int sum = 0;
  for each pc in partialCounts:
    sum += ParseInt(pc);
  Emit(word, AsString(sum));

Here is the self-explanatory Python code:

#!/usr/bin/python2.6
# Copyright [2011] Hemanth.HM
# Licensed under the Apache License, Version 2.0

import collections
import string

__authors__ = "Hemanth HM <hemanth.hm@gmail.com>"
__version__ = "1.0"

class MapReduce:
    """Linear implementation of MapReduce."""

    def map_it(self, lines):
        """Returns a list of lists of the form [word, 1] for each element."""
        return [[word, 1] for word in lines]

    def sort_it(self, wlist):
        """Returns a list of lists of the form [word, [partialcounts]]."""
        res = collections.defaultdict(list)
        map(lambda w: res[w[0]].append(w[1]), wlist)
        return res.items()

    def map_reduce(self, wlist):
        """Returns a dict with word : count mapping."""
        results = {}
        for res in self.sort_it(self.map_it(wlist)):
            results[res[0]] = sum(res[1])
        return results

class SlurpFile:
    """Simple class to get the file contents after filtering punctuation."""

    def __init__(self, path):
        self.fpath = path

    def get_contents(self):
        with open(self.fpath) as wfile:
            return ''.join(
                ch for ch in wfile.read() if ch not in set(string.punctuation)
            ).split()

mr = MapReduce()
slurp = SlurpFile('/tmp/t')
result = mr.map_reduce(slurp.get_contents())
print result

File contents: hemanth is testing ''''',,,,.;;;; hemanth
Output: {'hemanth': 2, 'is': 1, 'testing': 1}

Reference for the algorithm can be downloaded here.