Introducing Pyleus: An Open-source Framework for Building Storm Topologies in Pure Python

count_words.py

from collections import defaultdict

from pyleus.storm import SimpleBolt


class CountWordsBolt(SimpleBolt):

    def initialize(self):
        self.words = defaultdict(int)

    def process_tuple(self, tup):
        word, = tup.values

        self.words[word] += 1

        msg = "'{0}' has been seen {1} times\n".format(word, self.words[word])
        with open("/tmp/word_counts.txt", 'a') as f:
            f.write(msg)


if __name__ == '__main__':
    CountWordsBolt().run()

directory-tree

pyleus_topology.yaml
word_count/
    __init__.py
    line_spout.py
    split_words.py
    count_words.py

line_spout.py

import random

from pyleus.storm import Spout

LINES = """
Lorem ipsum dolor sit amet, consectetur
adipiscing elit. Curabitur pharetra ante eget
nunc blandit vestibulum. Curabitur tempus mi
...
vitae cursus leo, a congue justo.
""".strip().split('\n')


class LineSpout(Spout):

    OUTPUT_FIELDS = ["line"]

    def next_tuple(self):
        line = random.choice(LINES)
        tup = (line,)
        self.emit(tup)


if __name__ == '__main__':
    LineSpout().run()

pyleus_topology.yaml

name: word_count

topology:

    - spout:
        name: line-spout
        module: word_count.line_spout

    - bolt:
        name: split-words
        module: word_count.split_words
        groupings:
            - shuffle_grouping: line-spout

    - bolt:
        name: count-words
        module: word_count.count_words
        groupings:
            - fields_grouping:
                component: split-words
                fields:
                    - word