diff --git a/internal_use/CourseraSubmission.py b/internal_use/CourseraSubmission.py deleted file mode 100644 index c3fe8cd..0000000 --- a/internal_use/CourseraSubmission.py +++ /dev/null @@ -1,168 +0,0 @@ -#! /usr/bin/env python -u -# coding=utf-8 - -__author__ = 'Sayed Hadi Hashemi' - -import json -import urllib -import urllib2 -import hashlib -import email -import email.message -import email.encoders - - -class CourseraSubmission(object): - def __init__(self, **kwargs): - self.email = "" - self.password = "" - self.course_id = "" - self.part_ids = [] - self.part_names = [] - - self.__dict__.update(kwargs) - - def _login_prompt(self): - """Prompt the user for login credentials. Returns a tuple (login, password).""" - (login, password) = self._basic_prompt() - return login, password - - @staticmethod - def _basic_prompt(): - """Prompt the user for login credentials. Returns a tuple (login, password).""" - login = raw_input('Login (Email address): ') - password = raw_input('One-time Password (from the assignment page. This is NOT your own account\'s password): ') - return login, password - - def _auth_get_challenge(self, sid): - """Gets the challenge salt from the server. Returns (email,ch,state,ch_aux).""" - url = self._auth_get_challenge_url() - values = {'email_address': self.email, 'assignment_part_sid': sid, 'response_encoding': 'delim'} - data = urllib.urlencode(values) - req = urllib2.Request(url, data) - response = urllib2.urlopen(req) - text = response.read().strip() - - # text is of the form email|ch|signature - splits = text.split('|') - if len(splits) != 9: - print 'Badly formatted challenge response: %s' % text - return None - return splits[2], splits[4], splits[6], splits[8] - - def _auth_challenge_response(self, challenge): - sha1 = hashlib.sha1() - sha1.update("".join([challenge, self.password])) # hash the first elements - digest = sha1.hexdigest() - str_answer = '' - for i in range(0, len(digest)): - str_answer = str_answer + digest[i] - return str_answer - - def _auth_get_challenge_url(self): - """Returns the challenge url.""" - return "https://class.coursera.org/" + self.course_id + "/assignment/challenge" - - def _get_submission_url(self): - """Returns the submission url.""" - return "https://class.coursera.org/" + self.course_id + "/assignment/submit" - - def _submit_solution(self, ch_resp, sid, output, source, state, ch_aux): - """Submits a solution to the server. Returns (result, string).""" - source_64_msg = email.message.Message() - source_64_msg.set_payload(source) - email.encoders.encode_base64(source_64_msg) - - output_64_msg = email.message.Message() - output_64_msg.set_payload(output) - email.encoders.encode_base64(output_64_msg) - values = {'assignment_part_sid': sid, - 'email_address': self.email, - 'submission': output_64_msg.get_payload(), - 'submission_aux': source_64_msg.get_payload(), - 'challenge_response': ch_resp, - 'state': state - } - url = self._get_submission_url() - data = urllib.urlencode(values) - req = urllib2.Request(url, data) - response = urllib2.urlopen(req) - string = response.read().strip() - result = 0 - return result, string - - @staticmethod - def get_file_content(file_name): - with open(file_name, "r") as fp: - return fp.read() - - def submit(self): - print '\n== Connecting to Coursera ... ' - for part_index, part_id in enumerate(self.part_ids): - ret = self._auth_get_challenge(part_id) - if not ret: - print '\n!! Error: %s\n' % self.email - return False - - (login, ch, state, ch_aux) = ret - if (not self.email) or (not ch) or (not state): - print '\n!! Error: %s\n' % self.email - return - - if not self.is_enabled(part_index): - print '== (%s) %s' % (self.part_names[part_index], "Ignored. Result and/or Codes files are not exists.") - continue - ch_resp = self._auth_challenge_response(ch) - (result, string) = self._submit_solution(ch_resp, part_id, self.output(part_index), self.aux(part_index), - state, ch_aux) - - print '== (%s) %s' % (self.part_names[part_index], string.strip()) - if "We could not verify your username / password" in string: - return False - return True - - def aux(self, part_index): - return json.dumps({}) - - def output(self, part_index): - pass - - def run(self): - pass - - def is_enabled(self, part_index): - return True - - def init(self): - print '==\n== [sandbox] Submitting Solutions \n==' - - (self.email, self.password) = self._login_prompt() - if not self.email: - print '!! Submission Cancelled' - return False - - if len(self.part_ids) > 0: - sid = self.part_ids[0] - ret = self._auth_get_challenge(sid) - if not ret: - print '\n!! Error: %s\n' % self.email - return False - - (login, ch, state, ch_aux) = ret - if (not login) or (not ch) or (not state): - print '\n!! Error: %s\n' % login - return False - else: - return False - return True - - def make_sumbission(self): - if self.init(): - self.run() - while not self.submit(): - ret = raw_input('Try Again? (Y/N)') - if len(ret.strip()) > 0 and ret.strip().lower()[0] == 'y': - self.password = raw_input( - 'One-time Password (from the assignment page. This is NOT your own account\'s password): ') - else: - break diff --git a/internal_use/__init__.py b/internal_use/__init__.py deleted file mode 100644 index dc8ff74..0000000 --- a/internal_use/__init__.py +++ /dev/null @@ -1,4 +0,0 @@ -#! /usr/bin/env python -u -# coding=utf-8 - -__author__ = 'xl' diff --git a/internal_use/submit.py b/internal_use/submit.py deleted file mode 100755 index 2741389..0000000 --- a/internal_use/submit.py +++ /dev/null @@ -1,53 +0,0 @@ -#! /usr/bin/env python -u -# coding=utf-8 -import os - -__author__ = 'Sayed Hadi Hashemi' - -import json -from CourseraSubmission import CourseraSubmission - - -class MP3(CourseraSubmission): - def __init__(self): - super(CourseraSubmission, self).__init__() - dev = False - self.part_ids = ['mp3-part-a', 'mp3-part-b', 'mp3-part-c', 'mp3-part-d'] - for part_id in self.part_ids: - if dev: - part_id += "-dev" - - self.course_id = 'cloudapplications-001' - self.part_names = ["Top Word Finder Topology", "File Reader Spout", "Normalizer Bolt", "Top N Finder Bolt"] - self.files_results = ["output-part-a.txt", "output-part-b.txt", "output-part-c.txt", "output-part-d.txt"] - self.files_codes = [ - ["src/TopWordFinderTopologyPartA.java"], - ["src/FileReaderSpout.java", "src/TopWordFinderTopologyPartB.java"], - ["src/NormalizerBolt.java", "src/TopWordFinderTopologyPartC.java"], - ["src/TopNFinderBolt.java", "src/TopWordFinderTopologyPartD.java"] - ] - - @staticmethod - def get_file_content(file_name): - with open(file_name, "r") as fp: - return fp.read() - - def aux(self, part_index): - ret = {} - for filename in self.files_codes[part_index]: - ret[filename] = self.get_file_content(filename) - return json.dumps(ret) - - def output(self, part_index): - return self.get_file_content(self.files_results[part_index])[:400*1024] - - def is_enabled(self, part_index): - ret = True - ret = ret and os.path.exists(self.files_results[part_index]) - for filename in self.files_codes[part_index]: - ret = ret and os.path.exists(filename) - return ret - -if __name__ == "__main__": - mp1 = MP3() - mp1.make_sumbission() diff --git a/pom.xml b/pom.xml deleted file mode 100644 index 044d0c5..0000000 --- a/pom.xml +++ /dev/null @@ -1,45 +0,0 @@ - - 4.0.0 - - storm.example - storm-example - 0.0.1-SNAPSHOT - jar - - - UTF-8 - - - - - clojars.org - http://clojars.org/repo - - - - - - storm - storm - 0.9.0.1 - - provided - - - - - src - - - org.apache.maven.plugins - maven-compiler-plugin - - 1.6 - 1.6 - - - - - - diff --git a/python/config.json b/python/config.json new file mode 100644 index 0000000..118abeb --- /dev/null +++ b/python/config.json @@ -0,0 +1,23 @@ +{ + "serializer": "json", + "topology_specs": "topologies/", + "virtualenv_specs": "virtualenvs/", + "envs": { + "prod": { + "user": "", + "ssh_password": "", + "nimbus": "sandbox.hortonworks.com", + "use_ssh_for_nimbus": false, + "workers": [ + "sandbox.hortonworks.com" + ], + "log": { + "path": "/var/log/storm/streamparse", + "max_bytes": 1000000, + "backup_count": 10, + "level": "info" + }, + "virtualenv_root": "/data/virtualenvs/" + } + } +} diff --git a/python/fabfile.py b/python/fabfile.py new file mode 100644 index 0000000..e78d0bf --- /dev/null +++ b/python/fabfile.py @@ -0,0 +1,11 @@ +def pre_submit(topology_name, env_name, env_config): + """Override this function to perform custom actions prior to topology + submission. No SSH tunnels will be active when this function is called.""" + pass + + +def post_submit(topo_name, env_name, env_config): + """Override this function to perform custom actions after topology + submission. Note that the SSH tunnel to Nimbus will still be active + when this function is called.""" + pass diff --git a/python/postsetup.sh b/python/postsetup.sh new file mode 100644 index 0000000..4209018 --- /dev/null +++ b/python/postsetup.sh @@ -0,0 +1,28 @@ +pip install --upgrade pip +pip install --upgrade virtualenv + +mkdir -p $HOME/bin + +if [[ ":$PATH:" != *":$HOME/bin:"* ]]; then + PATH=$PATH:$HOME/bin + export PATH + echo "Added $HOME/bin to PATH" +fi + +wget -O $HOME/bin/lein https://raw.githubusercontent.com/technomancy/leiningen/stable/bin/lein +chmod a+x $HOME/bin/lein +echo "#!/bin/bash +export LEIN_ROOT=true" > /etc/profile.d/leinroot.sh +export LEIN_ROOT=true +lein + +mkdir -p /data/virtualenvs +touch /root/.ssh/config + +echo "/opt/rh/python27/root/usr/lib64/" >> /etc/ld.so.conf.d/x86_64-linux-gnu.conf +ldconfig + +mkdir -p /var/log/storm/streamparse +chown -R storm:hadoop /var/log/storm/streamparse + +pip install git+https://github.com/srujun/streamparse.git diff --git a/python/project.clj b/python/project.clj new file mode 100644 index 0000000..d26d008 --- /dev/null +++ b/python/project.clj @@ -0,0 +1,11 @@ +(defproject wordcount "0.0.1-SNAPSHOT" + :resource-paths ["_resources"] + :target-path "_build" + :min-lein-version "2.0.0" + :jvm-opts ["-client"] + :repositories { "HDP Releases" "http://repo.hortonworks.com/content/repositories/releases" } + :dependencies [[org.apache.storm/storm-core "0.10.0.2.3.2.0-2950"] + [org.apache.storm/flux-core "0.10.0.2.3.2.0-2950"]] + :jar-exclusions [#"log4j\.properties" #"org\.apache\.storm\.(?!flux)" #"trident" #"META-INF" #"meta-inf" #"\.yaml"] + :uberjar-exclusions [#"log4j\.properties" #"org\.apache\.storm\.(?!flux)" #"trident" #"META-INF" #"meta-inf" #"\.yaml"] + ) diff --git a/python/pysetup.sh b/python/pysetup.sh new file mode 100644 index 0000000..14650cf --- /dev/null +++ b/python/pysetup.sh @@ -0,0 +1,14 @@ +#!/bin/bash + +yum install -y nano centos-release-SCL zlib-devel \ +bzip2-devel openssl-devel ncurses-devel \ +sqlite-devel readline-devel tk-devel \ +gdbm-devel db4-devel libpcap-devel xz-devel \ +libpng-devel libjpg-devel atlas-devel + +yum groupinstall "Development tools" -y + +yum install -y python27 + +echo "#!/bin/bash +source /opt/rh/python27/enable" > /etc/profile.d/enablepython27.sh diff --git a/python/src/bolts/NormalizerBolt.py b/python/src/bolts/NormalizerBolt.py new file mode 100644 index 0000000..2f01025 --- /dev/null +++ b/python/src/bolts/NormalizerBolt.py @@ -0,0 +1,21 @@ +from streamparse import Bolt + +class NormalizerBolt(Bolt): + outputs = ['word'] + + def initialize(self, storm_conf, context): + self.common_words = [ + "the", "be", "a", "an", "and", "of", "to", "in", "am", + "is", "are", "at", "not", "that", "have", "i", "it", + "for", "on", "with", "he", "she", "as", "you", "do", + "this", "but", "his", "by", "from", "they", "we", "her", + "or", "will", "my", "one", "all", "s", "if", "any", "our", + "may", "your", "these", "d" , " ", "me" , "so" , "what" , "him" + ] + + def process(self, tup): + # TODO: + # Task 1: make the words all lower case + # Task 2: remove the common words + + pass diff --git a/python/src/bolts/SplitSentenceBolt.py b/python/src/bolts/SplitSentenceBolt.py new file mode 100644 index 0000000..c9b2d55 --- /dev/null +++ b/python/src/bolts/SplitSentenceBolt.py @@ -0,0 +1,9 @@ +from streamparse import Bolt + +class SplitSentenceBolt(Bolt): + outputs = ['word'] + + def process(self, tup): + sentence = tup.values[0] + for word in sentence.split(): + self.emit([word]) diff --git a/python/src/bolts/TopNFinderBolt.py b/python/src/bolts/TopNFinderBolt.py new file mode 100644 index 0000000..2d3036f --- /dev/null +++ b/python/src/bolts/TopNFinderBolt.py @@ -0,0 +1,30 @@ +from collections import Counter +import time + +from streamparse import Bolt + +class TopNFinderBolt(Bolt): + outputs = ['top-N'] + + def initialize(self, storm_conf, context): + self.top_words = Counter() + self.N = 10 + self.interval = 0.1 # 100 milliseconds + self.last_report = time.time() + + def process(self, tup): + # TODO: + # Task: keep track of the top N words + + + + # report the top N words periodically + if time.time() - self.last_report >= self.interval: + self.report() + + def report(self): + self.last_report = time.time() + + common_list = self.top_words.most_common(self.N) + self.logger.info('top-words = ' + str(common_list)) + self.emit([common_list]) diff --git a/python/src/bolts/WordCountBolt.py b/python/src/bolts/WordCountBolt.py new file mode 100644 index 0000000..9a192f0 --- /dev/null +++ b/python/src/bolts/WordCountBolt.py @@ -0,0 +1,15 @@ +from collections import Counter + +from streamparse import Bolt + +class WordCountBolt(Bolt): + outputs = ['word', 'count'] + + def initialize(self, storm_conf, context): + self.counter = Counter() + + def process(self, tup): + word = tup.values[0] + self.counter[word] += 1 + self.emit([word, self.counter[word]]) + self.logger.info("counted [{:,}] words [pid={}]".format(self.counter[word], self.pid)) diff --git a/python/src/bolts/__init__.py b/python/src/bolts/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/data.txt b/python/src/resources/data.txt similarity index 96% rename from data.txt rename to python/src/resources/data.txt index 8b70f8b..5ba49a3 100644 --- a/data.txt +++ b/python/src/resources/data.txt @@ -1,4 +1,4 @@ -***The Project Gutenberg's Etext of Shakespeare's First Folio*** +***The Project Gutenberg's Etext of Shakespeare's First Folio*** ********************The Tragedie of Macbeth********************* This is our 3rd edition of most of these plays. See the index. diff --git a/python/src/spouts/FileReaderSpout.py b/python/src/spouts/FileReaderSpout.py new file mode 100644 index 0000000..57e96b8 --- /dev/null +++ b/python/src/spouts/FileReaderSpout.py @@ -0,0 +1,27 @@ +import os +from os.path import join +from time import sleep + +from streamparse import Spout + +class FileReaderSpout(Spout): + outputs = ['word'] + + def initialize(self, stormconf, context): + datafile = join(os.getcwd(), stormconf['coursera.datafile']) + + # TODO: + # Task: Initialize the file reader + + + + def next_tuple(self): + # TODO: + # Task 1: read the next line and emit a tuple for it + # Task 2: don't forget to sleep for 1 second when the file is + # entirely read to prevent a busy-loop + + pass + + # NOTE: Streamparse does not have a close() function + # Closing the file should be handled in initialize() itself diff --git a/python/src/spouts/RandomSentenceSpout.py b/python/src/spouts/RandomSentenceSpout.py new file mode 100644 index 0000000..b1a3ff0 --- /dev/null +++ b/python/src/spouts/RandomSentenceSpout.py @@ -0,0 +1,21 @@ +from itertools import cycle +from time import sleep + +from streamparse import Spout + +class RandomSentenceSpout(Spout): + outputs = ['word'] + + def initialize(self, stormconf, context): + self.sentences = cycle([ + "the cow jumped over the moon", + "an apple a day keeps the doctor away", + "four score and seven years ago", + "snow white and the seven dwarfs", + "i am at two with nature" + ]) + + def next_tuple(self): + sleep(0.1) + sentence = next(self.sentences) + self.emit([sentence]) diff --git a/python/src/spouts/__init__.py b/python/src/spouts/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python/topologies/TopWordFinderTopologyPartA.py b/python/topologies/TopWordFinderTopologyPartA.py new file mode 100644 index 0000000..1f96c61 --- /dev/null +++ b/python/topologies/TopWordFinderTopologyPartA.py @@ -0,0 +1,17 @@ +from streamparse import Grouping, Topology + +from spouts.RandomSentenceSpout import RandomSentenceSpout +from bolts.SplitSentenceBolt import SplitSentenceBolt +from bolts.WordCountBolt import WordCountBolt + +class TopWordFinderTopologyPartA(Topology): + # TODO: + # Task: wire up the topology + # Make sure you use the following names for each component + # RandomSentenceSpout -> "spout" + # SplitSentenceBolt -> "split" + # WordCountBolt -> "count" + + + + # NOTE: will have to manually kill Topology after submission diff --git a/python/topologies/TopWordFinderTopologyPartB.py b/python/topologies/TopWordFinderTopologyPartB.py new file mode 100644 index 0000000..c4c6116 --- /dev/null +++ b/python/topologies/TopWordFinderTopologyPartB.py @@ -0,0 +1,19 @@ +from streamparse import Grouping, Topology + +from spouts.FileReaderSpout import FileReaderSpout +from bolts.SplitSentenceBolt import SplitSentenceBolt +from bolts.WordCountBolt import WordCountBolt + +class TopWordFinderTopologyPartA(Topology): + config = {'coursera.datafile': 'resources/data.txt'} + + # TODO: + # Task: wire up the topology + # Make sure you use the following names for each component + # FileReaderSpout -> "spout" + # SplitSentenceBolt -> "split" + # WordCountBolt -> "count" + + + + # NOTE: will have to manually kill Topology after submission diff --git a/python/topologies/TopWordFinderTopologyPartC.py b/python/topologies/TopWordFinderTopologyPartC.py new file mode 100644 index 0000000..db3048b --- /dev/null +++ b/python/topologies/TopWordFinderTopologyPartC.py @@ -0,0 +1,21 @@ +from streamparse import Grouping, Topology + +from spouts.FileReaderSpout import FileReaderSpout +from bolts.SplitSentenceBolt import SplitSentenceBolt +from bolts.NormalizerBolt import NormalizerBolt +from bolts.WordCountBolt import WordCountBolt + +class TopWordFinderTopologyPartC(Topology): + config = {'coursera.datafile': 'resources/data.txt'} + + # TODO: + # Task: wire up the topology + # Make sure you use the following names for each component + # FileReaderSpout -> "spout" + # SplitSentenceBolt -> "split" + # WordCountBolt -> "count" + # NormalizerBolt -> "normalize" + + + + # NOTE: will have to manually kill Topology after submission diff --git a/python/topologies/TopWordFinderTopologyPartD.py b/python/topologies/TopWordFinderTopologyPartD.py new file mode 100644 index 0000000..2805eaa --- /dev/null +++ b/python/topologies/TopWordFinderTopologyPartD.py @@ -0,0 +1,23 @@ +from streamparse import Grouping, Topology + +from spouts.FileReaderSpout import FileReaderSpout +from bolts.SplitSentenceBolt import SplitSentenceBolt +from bolts.NormalizerBolt import NormalizerBolt +from bolts.WordCountBolt import WordCountBolt +from bolts.TopNFinderBolt import TopNFinderBolt + +class TopWordFinderTopologyPartC(Topology): + config = {'coursera.datafile': 'resources/data.txt'} + + # TODO: + # Task: wire up the topology + # Make sure you use the following names for each component + # FileReaderSpout -> "spout" + # SplitSentenceBolt -> "split" + # WordCountBolt -> "count" + # NormalizerBolt -> "normalize" + # TopNFinderBolt -> "top-n" + + + + # NOTE: will have to manually kill Topology after submission diff --git a/python/virtualenvs/TopWordFinderTopologyPartA.txt b/python/virtualenvs/TopWordFinderTopologyPartA.txt new file mode 100644 index 0000000..9e1b9e3 --- /dev/null +++ b/python/virtualenvs/TopWordFinderTopologyPartA.txt @@ -0,0 +1,2 @@ +# streamparse # always required for streamparse projects +git+git://github.com/srujun/streamparse.git#egg=streamparse diff --git a/python/virtualenvs/TopWordFinderTopologyPartB.txt b/python/virtualenvs/TopWordFinderTopologyPartB.txt new file mode 100644 index 0000000..9e1b9e3 --- /dev/null +++ b/python/virtualenvs/TopWordFinderTopologyPartB.txt @@ -0,0 +1,2 @@ +# streamparse # always required for streamparse projects +git+git://github.com/srujun/streamparse.git#egg=streamparse diff --git a/python/virtualenvs/TopWordFinderTopologyPartC.txt b/python/virtualenvs/TopWordFinderTopologyPartC.txt new file mode 100644 index 0000000..9e1b9e3 --- /dev/null +++ b/python/virtualenvs/TopWordFinderTopologyPartC.txt @@ -0,0 +1,2 @@ +# streamparse # always required for streamparse projects +git+git://github.com/srujun/streamparse.git#egg=streamparse diff --git a/python/virtualenvs/TopWordFinderTopologyPartD.txt b/python/virtualenvs/TopWordFinderTopologyPartD.txt new file mode 100644 index 0000000..9e1b9e3 --- /dev/null +++ b/python/virtualenvs/TopWordFinderTopologyPartD.txt @@ -0,0 +1,2 @@ +# streamparse # always required for streamparse projects +git+git://github.com/srujun/streamparse.git#egg=streamparse diff --git a/settings.sh b/settings.sh deleted file mode 100755 index c679bce..0000000 --- a/settings.sh +++ /dev/null @@ -1,7 +0,0 @@ -#!/bin/bash -export XL_HOME=./internal_use - -export red=`tput setaf 1` -export green=`tput setaf 2` -export yellow=`tput setaf 3` -export reset=`tput sgr0` diff --git a/src/FileReaderSpout.java b/src/FileReaderSpout.java deleted file mode 100644 index 65a49f0..0000000 --- a/src/FileReaderSpout.java +++ /dev/null @@ -1,88 +0,0 @@ - -import java.io.BufferedReader; -import java.io.FileNotFoundException; -import java.io.FileReader; -import java.io.IOException; -import java.util.Map; - -import backtype.storm.spout.SpoutOutputCollector; -import backtype.storm.task.TopologyContext; -import backtype.storm.topology.IRichSpout; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Values; - -public class FileReaderSpout implements IRichSpout { - private SpoutOutputCollector _collector; - private TopologyContext context; - - - @Override - public void open(Map conf, TopologyContext context, - SpoutOutputCollector collector) { - - /* - ----------------------TODO----------------------- - Task: initialize the file reader - - - ------------------------------------------------- */ - - this.context = context; - this._collector = collector; - } - - @Override - public void nextTuple() { - - /* - ----------------------TODO----------------------- - Task: - 1. read the next line and emit a tuple for it - 2. don't forget to sleep when the file is entirely read to prevent a busy-loop - - ------------------------------------------------- */ - - - } - - @Override - public void declareOutputFields(OutputFieldsDeclarer declarer) { - - declarer.declare(new Fields("word")); - - } - - @Override - public void close() { - /* - ----------------------TODO----------------------- - Task: close the file - - - ------------------------------------------------- */ - - } - - - @Override - public void activate() { - } - - @Override - public void deactivate() { - } - - @Override - public void ack(Object msgId) { - } - - @Override - public void fail(Object msgId) { - } - - @Override - public Map getComponentConfiguration() { - return null; - } -} diff --git a/src/NormalizerBolt.java b/src/NormalizerBolt.java deleted file mode 100644 index bfd1305..0000000 --- a/src/NormalizerBolt.java +++ /dev/null @@ -1,41 +0,0 @@ -import backtype.storm.topology.BasicOutputCollector; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.topology.base.BaseBasicBolt; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Tuple; -import backtype.storm.tuple.Values; - -import java.util.Arrays; -import java.util.List; - -/** - * A bolt that normalizes the words, by removing common words and making them lower case. - */ -public class NormalizerBolt extends BaseBasicBolt { - private List commonWords = Arrays.asList("the", "be", "a", "an", "and", - "of", "to", "in", "am", "is", "are", "at", "not", "that", "have", "i", "it", - "for", "on", "with", "he", "she", "as", "you", "do", "this", "but", "his", - "by", "from", "they", "we", "her", "or", "will", "my", "one", "all", "s", "if", - "any", "our", "may", "your", "these", "d" , " ", "me" , "so" , "what" , "him" ); - - @Override - public void execute(Tuple tuple, BasicOutputCollector collector) { - - /* - ----------------------TODO----------------------- - Task: - 1. make the words all lower case - 2. remove the common words - - ------------------------------------------------- */ - - - } - - @Override - public void declareOutputFields(OutputFieldsDeclarer declarer) { - - declarer.declare(new Fields("word")); - - } -} diff --git a/src/RandomSentenceSpout.java b/src/RandomSentenceSpout.java deleted file mode 100644 index beab766..0000000 --- a/src/RandomSentenceSpout.java +++ /dev/null @@ -1,47 +0,0 @@ - - -import backtype.storm.spout.SpoutOutputCollector; -import backtype.storm.task.TopologyContext; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.topology.base.BaseRichSpout; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Values; -import backtype.storm.utils.Utils; - -import java.util.Map; -import java.util.Random; - -public class RandomSentenceSpout extends BaseRichSpout { - SpoutOutputCollector _collector; - Random _rand; - - - @Override - public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { - _collector = collector; - _rand = new Random(); - } - - @Override - public void nextTuple() { - Utils.sleep(100); - String[] sentences = new String[]{ "the cow jumped over the moon", "an apple a day keeps the doctor away", - "four score and seven years ago", "snow white and the seven dwarfs", "i am at two with nature" }; - String sentence = sentences[_rand.nextInt(sentences.length)]; - _collector.emit(new Values(sentence)); - } - - @Override - public void ack(Object id) { - } - - @Override - public void fail(Object id) { - } - - @Override - public void declareOutputFields(OutputFieldsDeclarer declarer) { - declarer.declare(new Fields("word")); - } - -} diff --git a/src/SplitSentenceBolt.java b/src/SplitSentenceBolt.java deleted file mode 100644 index 9b5194a..0000000 --- a/src/SplitSentenceBolt.java +++ /dev/null @@ -1,24 +0,0 @@ - -import backtype.storm.topology.BasicOutputCollector; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.topology.base.BaseBasicBolt; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Tuple; -import backtype.storm.tuple.Values; - -public class SplitSentenceBolt extends BaseBasicBolt { - -@Override - public void execute(Tuple tuple, BasicOutputCollector collector) { - String sentence = tuple.getString(0); - String[]words=sentence.split("[\\s~`!@#$%^&*(-)+=_:;'\",.<>?/\\\\0-9"+"\\]\\[\\}\\{]+"); - - for(String word:words){ - collector.emit(new Values(word)); - } - } - @Override - public void declareOutputFields(OutputFieldsDeclarer declarer) { - declarer.declare(new Fields("word")); - } - } diff --git a/src/TopNFinderBolt.java b/src/TopNFinderBolt.java deleted file mode 100644 index 1de5051..0000000 --- a/src/TopNFinderBolt.java +++ /dev/null @@ -1,61 +0,0 @@ -import backtype.storm.topology.BasicOutputCollector; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.topology.base.BaseBasicBolt; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Tuple; -import backtype.storm.tuple.Values; - -import java.util.HashMap; - -/** - * a bolt that finds the top n words. - */ -public class TopNFinderBolt extends BaseBasicBolt { - private HashMap currentTopWords = new HashMap(); - private int N; - - private long intervalToReport = 20; - private long lastReportTime = System.currentTimeMillis(); - - public TopNFinderBolt(int N) { - this.N = N; - } - - @Override - public void execute(Tuple tuple, BasicOutputCollector collector) { - /* - ----------------------TODO----------------------- - Task: keep track of the top N words - - - ------------------------------------------------- */ - - - //reports the top N words periodically - if (System.currentTimeMillis() - lastReportTime >= intervalToReport) { - collector.emit(new Values(printMap())); - lastReportTime = System.currentTimeMillis(); - } - } - - @Override - public void declareOutputFields(OutputFieldsDeclarer declarer) { - - declarer.declare(new Fields("top-N")); - - } - - public String printMap() { - StringBuilder stringBuilder = new StringBuilder(); - stringBuilder.append("top-words = [ "); - for (String word : currentTopWords.keySet()) { - stringBuilder.append("(" + word + " , " + currentTopWords.get(word) + ") , "); - } - int lastCommaIndex = stringBuilder.lastIndexOf(","); - stringBuilder.deleteCharAt(lastCommaIndex + 1); - stringBuilder.deleteCharAt(lastCommaIndex); - stringBuilder.append("]"); - return stringBuilder.toString(); - - } -} diff --git a/src/TopWordFinderTopologyPartA.java b/src/TopWordFinderTopologyPartA.java deleted file mode 100644 index 36256da..0000000 --- a/src/TopWordFinderTopologyPartA.java +++ /dev/null @@ -1,51 +0,0 @@ - -import backtype.storm.Config; -import backtype.storm.LocalCluster; -import backtype.storm.StormSubmitter; -import backtype.storm.topology.BasicOutputCollector; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.topology.TopologyBuilder; -import backtype.storm.topology.base.BaseBasicBolt; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Tuple; -import backtype.storm.tuple.Values; - -/** - * This topology counts the words from sentences emmited from a random sentence spout. - */ -public class TopWordFinderTopologyPartA { - - public static void main(String[] args) throws Exception { - - TopologyBuilder builder = new TopologyBuilder(); - - Config config = new Config(); - config.setDebug(true); - - - /* - ----------------------TODO----------------------- - Task: wire up the topology - - NOTE:make sure when connecting components together, using the functions setBolt(name,…) and setSpout(name,…), - you use the following names for each component: - - RandomSentanceSpout -> "spout" - SplitSentenceBolt -> "split" - WordCountBolt -> "count" - - - ------------------------------------------------- */ - - - config.setMaxTaskParallelism(3); - - LocalCluster cluster = new LocalCluster(); - cluster.submitTopology("word-count", config, builder.createTopology()); - - //wait for 60 seconds and then kill the topology - Thread.sleep(60 * 1000); - - cluster.shutdown(); - } -} diff --git a/src/TopWordFinderTopologyPartB.java b/src/TopWordFinderTopologyPartB.java deleted file mode 100644 index 9865a92..0000000 --- a/src/TopWordFinderTopologyPartB.java +++ /dev/null @@ -1,52 +0,0 @@ - -import backtype.storm.Config; -import backtype.storm.LocalCluster; -import backtype.storm.StormSubmitter; -import backtype.storm.topology.BasicOutputCollector; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.topology.TopologyBuilder; -import backtype.storm.topology.base.BaseBasicBolt; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Tuple; -import backtype.storm.tuple.Values; - -/** - * This topology reads a file and counts the words in that file - */ -public class TopWordFinderTopologyPartB { - - public static void main(String[] args) throws Exception { - - - TopologyBuilder builder = new TopologyBuilder(); - - Config config = new Config(); - config.setDebug(true); - - - /* - ----------------------TODO----------------------- - Task: wire up the topology - - NOTE:make sure when connecting components together, using the functions setBolt(name,…) and setSpout(name,…), - you use the following names for each component: - FileReaderSpout -> "spout" - SplitSentenceBolt -> "split" - WordCountBolt -> "count" - - - - ------------------------------------------------- */ - - - config.setMaxTaskParallelism(3); - - LocalCluster cluster = new LocalCluster(); - cluster.submitTopology("word-count", config, builder.createTopology()); - - //wait for 2 minutes and then kill the job - Thread.sleep( 2 * 60 * 1000); - - cluster.shutdown(); - } -} diff --git a/src/TopWordFinderTopologyPartC.java b/src/TopWordFinderTopologyPartC.java deleted file mode 100644 index 964a489..0000000 --- a/src/TopWordFinderTopologyPartC.java +++ /dev/null @@ -1,55 +0,0 @@ - -import backtype.storm.Config; -import backtype.storm.LocalCluster; -import backtype.storm.StormSubmitter; -import backtype.storm.topology.BasicOutputCollector; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.topology.TopologyBuilder; -import backtype.storm.topology.base.BaseBasicBolt; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Tuple; -import backtype.storm.tuple.Values; - -/** - * This topology reads a file, splits the senteces into words, normalizes the words such that all words are - * lower case and common words are removed, and then count the number of words. - */ -public class TopWordFinderTopologyPartC { - - public static void main(String[] args) throws Exception { - - - TopologyBuilder builder = new TopologyBuilder(); - - Config config = new Config(); - config.setDebug(true); - - - /* - ----------------------TODO----------------------- - Task: wire up the topology - - NOTE:make sure when connecting components together, using the functions setBolt(name,…) and setSpout(name,…), - you use the following names for each component: - - FileReaderSpout -> "spout" - SplitSentenceBolt -> "split" - WordCountBolt -> "count" - NormalizerBolt -> "normalize" - - - - ------------------------------------------------- */ - - - config.setMaxTaskParallelism(3); - - LocalCluster cluster = new LocalCluster(); - cluster.submitTopology("word-count", config, builder.createTopology()); - - //wait for 2 minutes then kill the job - Thread.sleep(2 * 60 * 1000); - - cluster.shutdown(); - } -} diff --git a/src/TopWordFinderTopologyPartD.java b/src/TopWordFinderTopologyPartD.java deleted file mode 100644 index f218f87..0000000 --- a/src/TopWordFinderTopologyPartD.java +++ /dev/null @@ -1,56 +0,0 @@ - -import backtype.storm.Config; -import backtype.storm.LocalCluster; -import backtype.storm.StormSubmitter; -import backtype.storm.topology.BasicOutputCollector; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.topology.TopologyBuilder; -import backtype.storm.topology.base.BaseBasicBolt; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Tuple; -import backtype.storm.tuple.Values; - -/** - * This topology reads a file and counts the words in that file, then finds the top N words. - */ -public class TopWordFinderTopologyPartD { - - private static final int N = 10; - - public static void main(String[] args) throws Exception { - - - TopologyBuilder builder = new TopologyBuilder(); - - Config config = new Config(); - config.setDebug(true); - - - /* - ----------------------TODO----------------------- - Task: wire up the topology - - NOTE:make sure when connecting components together, using the functions setBolt(name,…) and setSpout(name,…), - you use the following names for each component: - - FileReaderSpout -> "spout" - SplitSentenceBolt -> "split" - WordCountBolt -> "count" - NormalizerBolt -> "normalize" - TopNFinderBolt -> "top-n" - - - ------------------------------------------------- */ - - - config.setMaxTaskParallelism(3); - - LocalCluster cluster = new LocalCluster(); - cluster.submitTopology("word-count", config, builder.createTopology()); - - //wait for 2 minutes and then kill the job - Thread.sleep(2 * 60 * 1000); - - cluster.shutdown(); - } -} diff --git a/src/WordCountBolt.java b/src/WordCountBolt.java deleted file mode 100644 index e0c703b..0000000 --- a/src/WordCountBolt.java +++ /dev/null @@ -1,30 +0,0 @@ - -import backtype.storm.topology.BasicOutputCollector; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.topology.base.BaseBasicBolt; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Tuple; -import backtype.storm.tuple.Values; - -import java.util.HashMap; -import java.util.Map; - -public class WordCountBolt extends BaseBasicBolt { - Map counts = new HashMap(); - - @Override - public void execute(Tuple tuple, BasicOutputCollector collector) { - String word = tuple.getString(0); - Integer count = counts.get(word); - if (count == null) - count = 0; - count++; - counts.put(word, count); - collector.emit(new Values(word, count)); - } - - @Override - public void declareOutputFields(OutputFieldsDeclarer declarer) { - declarer.declare(new Fields("word", "count")); - } -} diff --git a/submit.sh b/submit.sh deleted file mode 100755 index 26bc21e..0000000 --- a/submit.sh +++ /dev/null @@ -1,8 +0,0 @@ -#!/bin/bash - -source settings.sh - -echo "${green}Processing the Results${reset}" -python $XL_HOME/submit.py - -echo "${green}Done${reset}"