<commit_before>#!/usr/bin/env python
# encoding: utf-8

'''
File: twitter_tz.py
Author: AFlock
Description: Given a database of tweets, aggregate number of tweets per timezone
Note: before running this job fill a mongodb with tweets by running :
curl https://stream.twitter.com/1/statuses/sample.json -u<user>:<pass> | mongoimport -c twitter


'''

from disco.core        import Job, result_iterator
from mongodb_io        import mongodb_input_stream
from app.MongoSplitter import calculate_splits

config = {
        "inputURI":"mongodb://localhost/twitter",
        "createInputSplits" : True,
        }

def map(tweet, params):
    yield tweet['user'].get('time_zone', "unlisted"), 1

def reduce(iter, params):
    from disco.util import kvgroup
    for zone, number in kvgroup(sorted(iter)):
        yield zone, sum(number)

if __name__ == '__main__':
    job = Job().run(input=calculate_splits(config) ,
            map = map,
            reduce = reduce,
            map_input_stream = mongodb_input_stream
            )

    for tz, count in result_iterator(job.wait(show=True)):
        print "TZ : %s :: Count : %s" % (tz, count)<commit_msg>twitter example pep8<commit_after>#!/usr/bin/env python
# encoding: utf-8

'''
File: twitter_tz.py
Author: AFlock
Description: Given a database of tweets, aggregate number of tweets per time
zone
Note: before running this job fill a mongodb with tweets by running :
curl https://stream.twitter.com/1/statuses/sample.json -u<user>:<pass> \
| mongoimport -c twitter


'''

from disco.core        import Job, result_iterator
from mongodb_io        import mongodb_input_stream
from app.MongoSplitter import calculate_splits

config = {
        "inputURI": "mongodb://localhost/twitter",
        "createInputSplits": True}


def map(tweet, params):
    yield tweet['user'].get('time_zone', "unlisted"), 1


def reduce(iter, params):
    from disco.util import kvgroup
    for zone, number in kvgroup(sorted(iter)):
        yield zone, sum(number)


if __name__ == '__main__':
    job = Job().run(input=calculate_splits(config),
            map=map,
            reduce=reduce,
            map_input_stream=mongodb_input_stream)

    for tz, count in result_iterator(job.wait(show=True)):
        print "TZ : %s :: Count : %s" % (tz, count)
{'fraction_non_alphanumeric': 0.11563731931668857, 'fraction_numerical': 0.0030661410424879547, 'mean_word_length': 3.3257575757575757, 'pattern_counts': {'":': 3, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5429083', 'n_tokens_mistral': 739, 'n_tokens_neox': 700, 'n_words': 200}