wordcount by sender

day5/mr_wc_by_sender.py

@@ -0,0 +1,18 @@
+import sys
+from mrjob.protocol import JSONValueProtocol
+from mrjob.job import MRJob
+from term_tools import get_terms
+
+class MRWordCount(MRJob):
+    INPUT_PROTOCOL = JSONValueProtocol
+    OUTPUT_PROTOCOL = JSONValueProtocol
+
+    def mapper(self, key, email):
+        for term in get_terms(email['text']):
+            yield {'term': term, 'sender': email['sender']}, 1
+
+    def reducer(self, word_sender, howmany):
+        yield None, {'term_sender': word_sender, 'count': sum(howmany)}
+
+if __name__ == '__main__':
+        MRWordCount.run()