storm post conversion

1 files changed, 237 insertions, 215 deletions

diff --git a/content/blog/Storm.markdown b/posts/storm.org
index 7223738..629f5a3 100644
--- a/content/blog/Storm.markdown
+++ b/posts/storm.org
@@ -1,177 +1,197 @@
----
-title: "Real-Time Streaming with Apache Storm and Apache Kafka"
-descritption: "Overview of Apache Storm and sample Twitter Sentiment Analysis"
-tags:
-  - "Apache Storm"
-  - "Apache Kafka"
-  - "Apache"
-  - "Java"
-  - "Sentiment Analysis"
-  - "Real-time Streaming"
-  - "ZData Inc."
-date: "2014-07-16"
-categories:
-  - "Apache"
-  - "Development"
-  - "Real-time Systems"
-slug: "real-time-streaming-storm-and-kafka"
----
-
-The following post is one in the series of real-time systems tangential
-to the Hadoop ecosystem.  First, exploring both Apache Storm and Apache
-Kafka as a part of a real-time processing engine. These two systems work
-together very well and make for an easy development experience while
-still being very performant.
-
-## About Kafka ##
-
-[Apache Kafka][3] is a message queue rethought as a distributed commit log. It
-follows the publish-subscribe messaging style, with speed and durability built
-in.
-
-Kafka uses Zookeeper to share and save state between brokers. Each broker
-maintains a set of partitions: primary and/ or secondary for each topic. A set
-of Kafka brokers working together will maintain a set of topics. Each topic has
-its partitions distributed over the participating Kafka brokers and, as of
+#+TITLE: Real-Time Streaming with Apache Storm and Apache Kafka
+#+DESCRIPTION: Overview of Apache Storm and sample Twitter Sentiment Analysis
+#+TAGS: Apache Storm
+#+TAGS: Apache Kafka
+#+TAGS: Apache
+#+TAGS: Java
+#+TAGS: Sentiment Analysis
+#+TAGS: Real-time streaming
+#+TAGS: zData Inc.
+#+DATE: 2014-07-16
+#+SLUG: real-time-streaming-storm-and-kafka
+#+LINK: kafka http://kafka.apache.org/
+#+LINK: kafka-benchmark http://engineering.linkedin.com/kafka/benchmarking-apache-kafka-2-million-writes-second-three-cheap-machines
+#+LINK: storm https://storm.apache.org/
+#+LINK: storm-multi-lingual https://storm.apache.org/about/multi-language.html
+#+LINK: storm-integrates https://storm.apache.org/about/integrates.html
+#+LINK: storm-common-patterns https://storm.apache.org/releases/current/Common-patterns.html
+#+LINK: docker http://www.docker.io/
+#+LINK: supervisord http://supervisord.org/
+#+LINK: storm-kafka-spout-github https://github.com/apache/incubator-storm/tree/master/external/storm-kafka
+#+LINK: wiki-stemming http://en.wikipedia.org/wiki/Stemming
+#+LINK: wiki-bag-of-words http://en.wikipedia.org/wiki/Bag-of-words_model
+#+LINK: storm-docs-stream-grouping http://storm.incubator.apache.org/documentation/Concepts.html#stream-groupings
+#+LINK: jackson-databing https://github.com/FasterXML/jackson-databind
+#+LINK: storm-trident-overview http://storm.incubator.apache.org/documentation/Trident-API-Overview.html
+#+LINK: wiki-srp http://en.wikipedia.org/wiki/Single_responsibility_principle
+#+LINK: storm-sample-project https://github.com/zdata-inc/StormSampleProject
+#+LINK: storm-incubation-proposal https://wiki.apache.org/incubator/StormProposal
+
+#+BEGIN_PREVIEW
+The following post is one in the series of real-time systems tangential to the
+Hadoop ecosystem.  First, exploring both Apache Storm and Apache Kafka as a
+part of a real-time processing engine.  These two systems work together very
+well and make for an easy development experience while still being very
+performant.
+#+END_PREVIEW
+
+** About Kafka
+
+[[kafka][Apache Kafka]] is a message queue rethought as a distributed commit
+log.  It follows the publish-subscribe messaging style, with speed and
+durability built in.
+
+Kafka uses Zookeeper to share and save state between brokers.  Each broker
+maintains a set of partitions: primary and/ or secondary for each topic.  A set
+of Kafka brokers working together will maintain a set of topics.  Each topic
+has its partitions distributed over the participating Kafka brokers and, as of
 Kafka version 0.8, the replication factor determines, intuitively, the number
 of times a partition is duplicated for fault tolerance.
 
 While many brokered message queue systems have the broker maintain the state of
-its consumers, Kafka does not. This frees up resources for the broker to ingest
-data faster. For more information about Kafka's performance see [Benchmarking
-Kafka][4].
+its consumers, Kafka does not.  This frees up resources for the broker to
+ingest data faster.  For more information about Kafka's performance see
+[[kafka-benchmark][Benchmarking Kafka]].
 
-### Initial Thoughts ###
+*** Initial Thoughts
 
-Kafka is a very promising project, with astounding throughput and one of
-the easiest pieces of software I have had the joy of installing and
-configuring. Although Kafka is not at the production 1.0 stable release yet,
-it's well on its way.
+Kafka is a very promising project, with astounding throughput and one of the
+easiest pieces of software I have had the joy of installing and configuring.
+Although Kafka is not at the production 1.0 stable release yet, it's well on
+its way.
 
-## About Storm ##
+** About Storm
 
-[Apache Storm][1], currently in incubation, is a real-time computational engine
-made available under the free and open-source Apache version 2.0 license. It
-runs continuously, consuming data from the configured sources (Spouts) and
-passes the data down the processing pipeline (Bolts). Combined, Spouts and
-Bolts make a Topology. A topology can be written in any language including any
-JVM based language, Python, Ruby, Perl, or, with some work, even C [[2][2]].
+[[storm][Apache Storm]], currently in incubation, is a real-time computational
+engine made available under the free and open-source Apache version 2.0
+license.  It runs continuously, consuming data from the configured sources
+(Spouts) and passes the data down the processing pipeline (Bolts).  Combined,
+Spouts and Bolts make a Topology.  A topology can be written in any language
+including any JVM based language, Python, Ruby, Perl, or, with some work, even
+C.  See the [[storm][Storm]] [[storm-multi-lingual][multi-lingual]]
+documentation.
 
-### Why Storm ###
+*** Why Storm
 
 Quoting from the project site:
 
-> Storm has many use cases: realtime analytics, online machine learning,
-> continuous computation, distributed RPC, ETL, and more. Storm is fast: a
-> benchmark clocked it at over a million tuples processed per second per node.
-> It is scalable, fault-tolerant, guarantees your data will be processed, and
-> is easy to set up and operate. [[1][1]]
+#+BEGIN_QUOTE
+  Storm has many use cases: realtime analytics, online machine learning,
+  continuous computation, distributed RPC, ETL, and more.  Storm is fast:
+  a benchmark clocked it at over a million tuples processed per second
+  per node.  It is scalable, fault-tolerant, guarantees your data will be
+  processed, and is easy to set up and operate.
+  [[storm][Storm Homepage]]
+#+END_QUOTE
 
-### Integration ###
+*** Integration
 
-Storm can integrate with any queuing and any database system. In fact, there
+Storm can integrate with any queuing and any database system.  In fact, there
 are already quite a few existing projects for use to integrate Storm with other
-projects, like Kestrel or Kafka[[5][5]].
+projects, like [[storm-integrates][kestrel or Kafka]].
 
-### Initial Thoughts ###
+*** Initial Thoughts
 
 I found Storm's verbiage around the computational pipeline to fit my mental
 model very well, thinking about streaming computational processes as directed
-acyclic graphs makes a lot of intuitive sense. That said, although I haven't
+acyclic graphs makes a lot of intuitive sense.  That said, although I haven't
 been developing against Storm for very long, I do find some integration tasks
-to be slightly awkward. For example, writing an HDFS file writer bolt
-requires some special considerations given bolt life cycles and HDFS writing
-patterns. This is really only a minor blemish however, as it only means the
-developers of Storm topologies have to understand the API more intimately;
-there are already common patterns emerging that should be adaptable to about
-any situation [[16][16]].
+to be slightly awkward.  For example, writing an HDFS file writer bolt requires
+some special considerations given bolt life cycles and HDFS writing patterns.
+This is really only a minor blemish however, as it only means the developers of
+Storm topologies have to understand the API more intimately; there are already
+common patterns emerging that should be adaptable to about any
+[[storm-common-patterns][situation]].
 
-## Test Project: Twitter Stream Sentiment Analysis ##
+** Test Project: Twitter Stream Sentiment Analysis
 
 To really give Storm a try, something a little more involved than just a simple
-word counter is needed. Therefore, I have put together a Twitter Sentiment
-Analysis topology. Though this is a good representative example of a more
+word counter is needed.  Therefore, I have put together a Twitter Sentiment
+Analysis topology.  Though this is a good representative example of a more
 complicated topology, the method used for actually scoring the Twitter data is
 overly simple.
 
-### Setup ###
+*** Setup
 
-The setup used for this demo is a 5 node Vagrant virtual cluster. Each node is
-running 64 bit CentOS 6.5, given 1 core, and 1024MB of RAM. Every node is
-running HDFS (datanode), Zookeeper, and Kafka. The first node, `node0`, is the
-namenode, and Nimbus -- Storm's master daemon. `node0` is also running a
-[Docker][7] container with a NodeJS application, part of the reporting process.
-The remaining nodes, `node[1-4]`, are Storm worker nodes. Storm, Kafka, and
-Zookeeper are all run under supervision via [Supervisord][6], so
-High-Availability is baked into this virtual cluster.
+The setup used for this demo is a 5 node Vagrant virtual cluster.  Each node is
+running 64 bit CentOS 6.5, given 1 core, and 1024MB of RAM.  Every node is
+running HDFS (datanode), Zookeeper, and Kafka.  The first node, ~node0~, is the
+namenode, and Nimbus -- Storm's master daemon.  ~node0~ is also running a
+[[docker][Docker]] container with a NodeJS application, part of the reporting
+process.  The remaining nodes, ~node[1-4]~, are Storm worker nodes.  Storm,
+Kafka, and Zookeeper are all run under supervision via
+[[supervisord][Supervisord]], so High-Availability is baked into this virtual
+cluster.
 
-### Overview ###
+*** Overview
 
-{{< figure src="/media/SentimentAnalysisTopology.png"
-    alt="Sentiment Analysis Topology">}}
+[[file:/media/SentimentAnalysisTopology.png]]
 
 I wrote a simple Kafka producer that reads files off disk and sends them to the
-Kafka cluster. This is how we feed the whole system and is used in lieu of
+Kafka cluster.  This is how we feed the whole system and is used in lieu of
 opening a stream to Twitter.
 
-#### Spout ####
+**** Spout
 
-The orange node from the picture is our [`KafkaSpout`][8] that will be
-consuming incoming messages from the Kafka brokers.
+The orange node from the picture is our
+[[storm-kafka-spout-github][~KafkaSpout~]] that will be consuming incoming
+messages from the Kafka brokers.
 
-#### Twitter Data JSON Parsing ####
+**** Twitter Data JSON Parsing
 
-The first bolt, `2` in the image, attempts to parse the Twitter JSON data and
-emits `tweet_id` and `tweet_text`. This implementation only processes English
-tweets. If the topology were to be more ambitious, it may pass the language
+The first bolt, ~2~ in the image, attempts to parse the Twitter JSON data and
+emits =tweet_id= and =tweet_text=.  This implementation only processes English
+tweets.  If the topology were to be more ambitious, it may pass the language
 code down and create different scoring bolts for each language.
 
-#### Filtering and Stemming ####
+**** Filtering and Stemming
 
-This next bolt, `3`, performs first-round data sanitization. That is, it
+This next bolt, ~3~, performs first-round data sanitization.  That is, it
 removes all non-alpha characters.
 
-Following, the next round of data cleansing, `4`, is to remove common words
-to reduce noise for the classifiers. These common words are usually referred to
-as _stop words_. [_Stemming_][15], or considering suffixes to match the root,
-could also be performed here, or in another, later bolt.
+Following, the next round of data cleansing, ~4~, is to remove common words to
+reduce noise for the classifiers.  These common words are usually referred to
+as /stop words/.  [[wiki-stemming][/Stemming/]], or considering suffixes to
+match the root, could also be performed here, or in another, later bolt.
 
-`4`, when finished, will then broadcast the data to both of the classifiers.
+~4~, when finished, will then broadcast the data to both of the classifiers.
 
-#### Classifiers ####
+**** Classifiers
 
-Each classifier is defined by its own bolt. One classifier for the positive
-class, `5`, and another for the negative class,`6`.
+Each classifier is defined by its own bolt.  One classifier for the positive
+class, ~5~, and another for the negative class, ~6~.
 
-The implementation of the classifiers follows the [Bag-of-words][12] model.
+The implementation of the classifiers follows the
+[[wiki-bag-of-words][Bag-of-words]] model.
 
-#### Join and Scoring ####
+**** Join and Scoring
 
-Next, bolt `7` joins the scores from the two previous classifiers. The
-implementation of this bolt isn't perfect. For example, message transaction is
+Next, bolt ~7~ joins the scores from the two previous classifiers.  The
+implementation of this bolt isn't perfect.  For example, message transaction is
 not guaranteed: if one-side of the join fails, neither side is notified.
 
-To finish up the scoring, bolt `8` compares the scores from the classifiers and
+To finish up the scoring, bolt ~8~ compares the scores from the classifiers and
 declares the class accordingly.
 
-#### Reporting: HDFS and HTTP POST ####
+**** Reporting: HDFS and HTTP POST
 
 Finally, the scoring bolt broadcasts off the results to a HDFS file writer
-bolt, `9`, and a NodeJS notifier bolt, `10`. The HDFS bolt fills a list until
-it has 1000 records in it and then spools to disk. Of course, like the join
+bolt, ~9~, and a NodeJS notifier bolt, ~10~.  The HDFS bolt fills a list until
+it has 1000 records in it and then spools to disk.  Of course, like the join
 bolt, this could be better implemented to fail input tuples if the bolt fails
-to write to disk or have a timeout to write to disk after a certain
-period of time. The NodeJs notifier bolt, on the other hand, sends a HTTP POST
-each time a record is received. This could be batched as well, but again, this
-is for demonstration purposes only.
+to write to disk or have a timeout to write to disk after a certain period of
+time.  The NodeJs notifier bolt, on the other hand, sends a HTTP POST each time
+a record is received.  This could be batched as well, but again, this is for
+demonstration purposes only.
 
-### Implementing the Kafka Producer ###
+*** Implementing the Kafka Producer
 
 Here, the main portions of the code for the Kafka producer that was used to
 test our cluster are defined.
 
 In the main class, we setup the data pipes and threads:
 
+#+BEGIN_SRC java
     LOGGER.debug("Setting up streams");
     PipedInputStream send = new PipedInputStream(BUFFER_LEN);
     PipedOutputStream input = new PipedOutputStream(send);
@@ -191,9 +211,11 @@ In the main class, we setup the data pipes and threads:
 
     LOGGER.debug("Joining");
     kafka.join();
+#+END_SRC
 
-The `BufferedFileReader` in its own thread reads off the data from disk:
+The ~BufferedFileReader~ in its own thread reads off the data from disk:
 
+#+BEGIN_SRC java
     rd = new BufferedReader(new FileReader(this.fileToRead));
     wd = new BufferedWriter(new OutputStreamWriter(this.outputStream, ENC));
     int b = -1;
@@ -201,9 +223,11 @@ The `BufferedFileReader` in its own thread reads off the data from disk:
     {
         wd.write(b);
     }
+#+END_SRC
 
-Finally, the `KafkaProducer` sends asynchronous messages to the Kafka Cluster:
+Finally, the ~KafkaProducer~ sends asynchronous messages to the Kafka Cluster:
 
+#+BEGIN_SRC java
     rd = new BufferedReader(new InputStreamReader(this.inputStream, ENC));
     String line = null;
     producer = new Producer<Integer, String>(conf);
@@ -211,18 +235,23 @@ Finally, the `KafkaProducer` sends asynchronous messages to the Kafka Cluster:
     {
         producer.send(new KeyedMessage<Integer, String>(this.topic, line));
     }
+#+END_SRC
 
 Doing these operations on separate threads gives us the benefit of having disk
 reads not block the Kafka producer or vice-versa, enabling maximum performance
 tunable by the size of the buffer.
 
-### Implementing the Storm Topology ###
+*** Implementing the Storm Topology
 
-#### Topology Definition ####
+**** Topology Definition
+     :PROPERTIES:
+     :CUSTOM_ID: topology-definition
+     :END:
 
-Moving onward to Storm, here we define the topology and how each bolt will be
-talking to each other:
+Moving onward to Storm, here we define the topology and how each bolt
+will be talking to each other:
 
+#+BEGIN_SRC java
     TopologyBuilder topology = new TopologyBuilder();
 
     topology.setSpout("kafka_spout", new KafkaSpout(kafkaConf), 4);
@@ -252,17 +281,20 @@ talking to each other:
             .shuffleGrouping("score");
     topology.setBolt("nodejs", new NodeNotifierBolt(), 4)
             .shuffleGrouping("score");
+#+END_SRC
 
 Notably, the data is shuffled to each bolt until except when joining, as it's
 very important that the same tweets are given to the same instance of the
-joining bolt. To read more about stream groupings, visit the [Storm
-documentation][17].
+joining bolt.  To read more about stream groupings, visit the
+[[storm-docs-stream-grouping][Storm documentation]].
 
-#### Twitter Data Filter / Parser ####
+**** Twitter Data Filter / Parser
 
 Parsing the Twitter JSON data is one of the first things that needs to be done.
-This is accomplished with the use of the [JacksonXML Databind][11] library.
+This is accomplished with the use of the [[jackson-databind][JacksonXML
+Databind]] library.
 
+#+BEGIN_SRC java
     JsonNode root = mapper.readValue(json, JsonNode.class);
     long id;
     String text;
@@ -280,30 +312,34 @@ This is accomplished with the use of the [JacksonXML Databind][11] library.
     }
     else
         LOGGER.debug("Ignoring non-english tweet");
+#+END_SRC
 
-As mentioned before, `tweet_id` and `tweet_text` are the only values being
+As mentioned before, ~tweet_id~ and ~tweet_text~ are the only values being
 emitted.
 
-This is just using the basic `ObjectMapper` class from the Jackson Databind
+This is just using the basic ~ObjectMapper~ class from the Jackson Databind
 library to map the simple JSON object provided by the Twitter Streaming API to
-a `JsonNode`. The language code is first tested to be English, as the topology
-does not support non-English tweets. The new tuple is emitted down the Storm
-pipeline after ensuring the existence of the desired data, namely, `tweet_id`,
-and `tweet_text`.
+a ~JsonNode~.  The language code is first tested to be English, as the topology
+does not support non-English tweets.  The new tuple is emitted down the Storm
+pipeline after ensuring the existence of the desired data, namely, ~tweet_id~,
+and ~tweet_text~.
 
-#### Text Filtering ####
+**** Text Filtering
 
 As previously mentioned, punctuation and other symbols are removed because they
 are just noise to the classifiers:
 
+#+BEGIN_SRC java
     Long id = input.getLong(input.fieldIndex("tweet_id"));
     String text = input.getString(input.fieldIndex("tweet_text"));
     text = text.replaceAll("[^a-zA-Z\\s]", "").trim().toLowerCase();
     collector.emit(new Values(id, text));
+#+END_SRC
 
-_Very_ common words are also removed because they are similarly noisy to the
+/Very/ common words are also removed because they are similarly noisy to the
 classifiers:
 
+#+BEGIN_SRC java
     Long id = input.getLong(input.fieldIndex("tweet_id"));
     String text = input.getString(input.fieldIndex("tweet_text"));
     List<String> stopWords = StopWords.getWords();
@@ -312,18 +348,20 @@ classifiers:
         text = text.replaceAll(word, "");
     }
     collector.emit(new Values(id, text));
+#+END_SRC
 
-Here the `StopWords` class is a singleton holding the list of words we
-wish to omit.
+Here the ~StopWords~ class is a singleton holding the list of words we wish to
+omit.
 
-#### Positive and Negative Scoring ####
+**** Positive and Negative Scoring
 
-Since the approach for scoring is based on a very limited [Bag-of-words][12]
-model, Each class is put into its own bolt; this also contrives the need for a
-join later.
+Since the approach for scoring is based on a very limited
+[[wiki-bag-of-words][Bag-of-words]] model, Each class is put into its own bolt;
+this also contrives the need for a join later.
 
 Positive Scoring:
 
+#+BEGIN_SRC java
     Long id = input.getLong(input.fieldIndex("tweet_id"));
     String text = input.getString(input.fieldIndex("tweet_text"));
     Set<String> posWords = PositiveWords.getWords();
@@ -336,9 +374,11 @@ Positive Scoring:
             numPosWords++;
     }
     collector.emit(new Values(id, (float) numPosWords / numWords, text));
+#+END_SRC
 
 Negative Scoring:
 
+#+BEGIN_SRC java
     Long id = input.getLong(input.fieldIndex("tweet_id"));
     String text = input.getString(input.fieldIndex("tweet_text"));
     Set<String> negWords = NegativeWords.getWords();
@@ -351,21 +391,24 @@ Negative Scoring:
             numNegWords++;
     }
     collector.emit(new Values(id, (float)numNegWords / numWords, text));
+#+END_SRC
 
-Similar to `StopWords`, `PositiveWords` and `NegativeWords` are both singletons
+Similar to ~StopWords~, ~PositiveWords~ and ~NegativeWords~ are both singletons
 maintaining lists of positive and negative words, respectively.
 
-#### Joining Scores ####
+**** Joining Scores
 
 Joining in Storm isn't the most straight forward process to implement, although
-the process may be made simpler with the use of the [Trident API][13]. However,
-to get a feel for what to do without Trident and as an Academic exercise, the
-join is not implemented with the Trident API. On related note, this join
-isn't perfect! It ignores a couple of issues, namely, it does not correctly
-fail a tuple on a one-sided join (when tweets are received twice from the same
-scoring bolt) and doesn't timeout tweets if they are left in the queue for too
-long.  With this in mind, here is our join:
-
+the process may be made simpler with the use of the
+[[storm-trident-overview][Trident API]].  However, to get a feel for what to do
+without Trident and as an Academic exercise, the join is not implemented with
+the Trident API.  On related note, this join isn't perfect! It ignores a couple
+of issues, namely, it does not correctly fail a tuple on a one-sided join (when
+tweets are received twice from the same scoring bolt) and doesn't timeout
+tweets if they are left in the queue for too long.  With this in mind, here is
+our join:
+
+#+BEGIN_SRC java
     Long id = input.getLong(input.fieldIndex("tweet_id"));
     String text = input.getString(input.fieldIndex("tweet_text"));
     if (input.contains("pos_score"))
@@ -406,9 +449,11 @@ long.  With this in mind, here is our join:
                 id,
                 new Triple<String, Float, String>("neg", neg, text));
     }
+#+END_SRC
 
-Where `emit` is defined simply by:
+Where ~emit~ is defined simply by:
 
+#+BEGIN_SRC java
     private void emit(
         BasicOutputCollector collector,
         Long id,
@@ -419,28 +464,32 @@ Where `emit` is defined simply by:
         collector.emit(new Values(id, pos, neg, text));
         this.tweets.remove(id);
     }
+#+END_SRC
 
-#### Deciding the Winning Class ####
+**** Deciding the Winning Class
 
-To ensure the [Single responsibility principle][14] is not violated, joining
-and scoring are split into separate bolts, though the class of the tweet could
-certainly be decided at the time of joining.
+To ensure the [[wiki-srp][Single responsibility principle]] is not violated,
+joining and scoring are split into separate bolts, though the class of the
+tweet could certainly be decided at the time of joining.
 
+#+BEGIN_SRC java
     Long id = tuple.getLong(tuple.fieldIndex("tweet_id"));
     String text = tuple.getString(tuple.fieldIndex("tweet_text"));
     Float pos = tuple.getFloat(tuple.fieldIndex("pos_score"));
     Float neg = tuple.getFloat(tuple.fieldIndex("neg_score"));
     String score = pos > neg ? "positive" : "negative";
     collector.emit(new Values(id, text, pos, neg, score));
+#+END_SRC
 
 This decider will prefer negative scores, so if there is a tie, it's
 automatically handed to the negative class.
 
-#### Report the Results ####
+**** Report the Results
 
-Finally, there are two bolts that will yield results: one that writes
-data to HDFS, and another to send the data to a web server.
+Finally, there are two bolts that will yield results: one that writes data to
+HDFS, and another to send the data to a web server.
 
+#+BEGIN_SRC java
     Long id = input.getLong(input.fieldIndex("tweet_id"));
     String tweet = input.getString(input.fieldIndex("tweet_text"));
     Float pos = input.getFloat(input.fieldIndex("pos_score"));
@@ -454,9 +503,11 @@ data to HDFS, and another to send the data to a web server.
         writeToHDFS();
         this.tweet_scores = new ArrayList<String>(1000);
     }
+#+END_SRC
 
-Where `writeToHDFS` is primarily given by:
+Where ~writeToHDFS~ is primarily given by:
 
+#+BEGIN_SRC java
     Configuration conf = new Configuration();
     conf.addResource(new Path("/opt/hadoop/etc/hadoop/core-site.xml"));
     conf.addResource(new Path("/opt/hadoop/etc/hadoop/hdfs-site.xml"));
@@ -472,9 +523,11 @@ Where `writeToHDFS` is primarily given by:
     {
         wd.write(tweet_score);
     }
+#+END_SRC
 
-And our `HTTP POST` to a web server:
+And our ~HTTP POST~ to a web server:
 
+#+BEGIN_SRC java
     Long id = input.getLong(input.fieldIndex("tweet_id"));
     String tweet = input.getString(input.fieldIndex("tweet_text"));
     Float pos = input.getFloat(input.fieldIndex("pos_score"));
@@ -501,19 +554,21 @@ And our `HTTP POST` to a web server:
         LOGGER.trace(null, ex);
         reconnect();
     }
+#+END_SRC
 
-There are some faults to point out with both of these bolts. Namely, the HDFS
+There are some faults to point out with both of these bolts.  Namely, the HDFS
 bolt tries to batch the writes into 1000 tweets, but does not keep track of the
-tuples nor does it time out tuples or flush them at some interval. That is, if
+tuples nor does it time out tuples or flush them at some interval.  That is, if
 a write fails or if the queue sits idle for too long, the topology is not
-notified and can't resend the tuples. Similarly, the `HTTP POST`, does not
+notified and can't resend the tuples.  Similarly, the ~HTTP POST~, does not
 batch and sends each POST synchronously causing the bolt to block for each
-message. This can be alleviated with more instances of this bolt and more web
+message.  This can be alleviated with more instances of this bolt and more web
 servers to handle the increase in posts, and/ or a better batching process.
 
-## Summary ##
+** Summary
 
-The full source of this test application can be found on [Github][9].
+The full source of this test application can be found on
+[[storm-sample-project][Github]].
 
 Apache Storm and Apache Kafka both have great potential in the real-time
 streaming market and have so far proven themselves to be very capable systems
@@ -522,72 +577,39 @@ for performing real-time analytics.
 Stay tuned, as the next post in this series will be an overview look at
 Streaming with Apache Spark.
 
-## Related Links / References ##
-
-[1]: http://storm.incubator.apache.org/
-
-*   [Apache Storm Project Page][1]
-
-[2]: http://storm.incubator.apache.org/about/multi-language.html
-
-*   [Storm Multi-Language Documentation][2]
-
-[3]: http://kafka.apache.org/
-
-*   [Apache Kafka Project Page][3]
-
-[4]: http://engineering.linkedin.com/kafka/benchmarking-apache-kafka-2-million-writes-second-three-cheap-machines
-
-*   [LinkedIn Kafka Benchmarking: 2 million writes per second][4]
-
-[5]: http://storm.incubator.apache.org/about/integrates.html
-
-*   [Storm Integration Documentation][5]
-
-[6]: http://supervisord.org/
-
-*   [Supervisord Project Page][6]
-
-[7]: http://www.docker.io/
-
-*   [Docker IO Project Page][7]
-
-[8]: https://github.com/apache/incubator-storm/tree/master/external/storm-kafka
-
-*   [Storm-Kafka Source][8]
-
-[9]: https://github.com/zdata-inc/StormSampleProject
+** Related Links / References
 
-*   [Full Source of Test Project][9]
+-  [[storm][Apache Storm Project Page]]
 
-[10]: https://wiki.apache.org/incubator/StormProposal
+-  [[storm-multi-lingual][Storm Multi-Language Documentation]]
 
-*   [Apache Storm Incubation Proposal][10]
+-  [[kafka][Apache Kafka Project Page]]
 
-[11]: https://github.com/FasterXML/jackson-databind
+-  [[kafka-benchmark][LinkedIn Kafka Benchmarking: 2 million writes per
+  second]]
 
-*   [Jackson Databind Project Bag][11]
+-  [[storm-integrates][Storm Integration Documentation]]
 
-[12]: http://en.wikipedia.org/wiki/Bag-of-words_model
+-  [[supervisord][Supervisord Project Page]]
 
-*   [Wikipedia: Bag of words][12]
+-  [[docker][Docker IO Project Page]]
 
-[13]: http://storm.incubator.apache.org/documentation/Trident-API-Overview.html
+-  [[storm-kafka-extern-github][Storm-Kafka Source]]
 
-*   [Storm Trident API Overview][13]
+-  [[storm-sample-project][Full Source of Test Project]]
 
-[14]: http://en.wikipedia.org/wiki/Single_responsibility_principle
+-  [[storm-incubation-proposal][Apache Storm Incubation Proposal]]
 
-*   [Wikipedia: Single responsibility principle][14]
+-  [[jackson-databind][Jackson Databind Project Bag]]
 
-[15]: http://en.wikipedia.org/wiki/Stemming
+-  [[wiki-bag-of-words][Wikipedia: Bag of words]]
 
-*   [Wikipedia: Stemming][15]
+-  [[storm-trident-overview][Storm Trident API Overview]]
 
-[16]: http://storm.incubator.apache.org/documentation/Common-patterns.html
+-  [[wiki-srp][Wikipedia: Single responsibility principle]]
 
-*   [Storm Documentation: Common Patterns][16]
+-  [[wiki-stemming][Wikipedia: Stemming]]
 
-[17]: http://storm.incubator.apache.org/documentation/Concepts.html#stream-groupings
+-  [[storm-common-patterns][Storm Documentation: Common Patterns]]
 
-*   [Stream Groupings][17]
+-  [[storm-docs-stream-grouping][Stream Groupings]]