count_bigrams.py

@router.node(('bigram',))
def bigrams(msg):
    'get bigrams from a document'
    prev = None
    for word in msg.document.strip().split(' '):
        yield (prev, word)
        prev = word

@router.node(('bigram', 'frequency'), 'bigrams')
def count_bigrams(msg):
    'count bigrams'
    count = redis.incrby(' '.join(msg.bigram), 1)
    return bigram, count

filtering.py

@router.node(('tweet',), 'some_other_node')
def filter_tweets(msg):
    'filter tweets based on content'
    if BAD_WORD in msg.tweet:
        return NoResult
    
    return msg.tweet

generator_example.py

@router.node(('word',))
def get_words(msg):
    'get the words in a document'
    for word in msg.document.strip().split(' '):
        if word:
            yield word

return_example.py

@router.node(('word', 'count'), 'package.other_func')
def count_words(msg):
    'get the count of a word'
    count = redis_conn.incrby(msg.word, 1)
    return msg.word, count

router.py

from emit import Router
router = Router()

# or

from emit.router.rq import RQRouter
from redis import Redis

router = RQRouter(Redis())

sentiment.py

@router.node(('sentence',))
def sentences(msg):
    'naively split a document into sentences'
    key = '.  ' # dot space space
    for sentence in msg.document.strip().split(key):
        yield sentence

@router.node(('sentence', 'score'), 'sentences')
def sentiment(msg):
    'analyze sentiment'
    # NLTK magic here
    return msg.sentence, sentiment