#+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
[[kafka-benchmark][Benchmarking Kafka]].

*** 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.

** About Storm

[[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

Quoting from the project site:

#+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

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 [[storm-integrates][kestrel or Kafka]].

*** 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
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
[[storm-common-patterns][situation]].

** 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
complicated topology, the method used for actually scoring the Twitter data is
overly simple.

*** 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][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

[[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
opening a stream to Twitter.

**** Spout

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

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

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/.  [[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.

**** Classifiers

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
[[wiki-bag-of-words][Bag-of-words]] model.

**** 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
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
declares the class accordingly.

**** 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, 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.

*** 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);

    LOGGER.debug("Setting up connections");
    LOGGER.debug("Setting up file reader");
    BufferedFileReader reader = new BufferedFileReader(filename, input);
    LOGGER.debug("Setting up kafka producer");
    KafkaProducer kafkaProducer = new KafkaProducer(topic, send);

    LOGGER.debug("Spinning up threads");
    Thread source = new Thread(reader);
    Thread kafka = new Thread(kafkaProducer);

    source.start();
    kafka.start();

    LOGGER.debug("Joining");
    kafka.join();
#+END_SRC

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;
    while ((b = rd.read()) != -1)
    {
        wd.write(b);
    }
#+END_SRC

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);
    while ((line = rd.readLine()) != null)
    {
        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

**** 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:

#+BEGIN_SRC java
    TopologyBuilder topology = new TopologyBuilder();

    topology.setSpout("kafka_spout", new KafkaSpout(kafkaConf), 4);

    topology.setBolt("twitter_filter", new TwitterFilterBolt(), 4)
            .shuffleGrouping("kafka_spout");

    topology.setBolt("text_filter", new TextFilterBolt(), 4)
            .shuffleGrouping("twitter_filter");

    topology.setBolt("stemming", new StemmingBolt(), 4)
            .shuffleGrouping("text_filter");

    topology.setBolt("positive", new PositiveSentimentBolt(), 4)
            .shuffleGrouping("stemming");
    topology.setBolt("negative", new NegativeSentimentBolt(), 4)
            .shuffleGrouping("stemming");

    topology.setBolt("join", new JoinSentimentsBolt(), 4)
            .fieldsGrouping("positive", new Fields("tweet_id"))
            .fieldsGrouping("negative", new Fields("tweet_id"));

    topology.setBolt("score", new SentimentScoringBolt(), 4)
            .shuffleGrouping("join");

    topology.setBolt("hdfs", new HDFSBolt(), 4)
            .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-docs-stream-grouping][Storm documentation]].

**** 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 [[jackson-databind][JacksonXML
Databind]] library.

#+BEGIN_SRC java
    JsonNode root = mapper.readValue(json, JsonNode.class);
    long id;
    String text;
    if (root.get("lang") != null &&
        "en".equals(root.get("lang").textValue()))
    {
        if (root.get("id") != null && root.get("text") != null)
        {
            id = root.get("id").longValue();
            text = root.get("text").textValue();
            collector.emit(new Values(id, text));
        }
        else
            LOGGER.debug("tweet id and/ or text was null");
    }
    else
        LOGGER.debug("Ignoring non-english tweet");
#+END_SRC

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
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~.

**** 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
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();
    for (String word : stopWords)
    {
        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.

**** Positive and Negative Scoring

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();
    String[] words = text.split(" ");
    int numWords = words.length;
    int numPosWords = 0;
    for (String word : words)
    {
        if (posWords.contains(word))
            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();
    String[] words = text.split(" ");
    int numWords = words.length;
    int numNegWords = 0;
    for (String word : words)
    {
        if (negWords.contains(word))
            numNegWords++;
    }
    collector.emit(new Values(id, (float)numNegWords / numWords, text));
#+END_SRC

Similar to ~StopWords~, ~PositiveWords~ and ~NegativeWords~ are both singletons
maintaining lists of positive and negative words, respectively.

**** 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
[[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"))
    {
        Float pos = input.getFloat(input.fieldIndex("pos_score"));
        if (this.tweets.containsKey(id))
        {
            Triple<String, Float, String> triple = this.tweets.get(id);
            if ("neg".equals(triple.getCar()))
                emit(collector, id, triple.getCaar(), pos, triple.getCdr());
            else
            {
                LOGGER.warn("one sided join attempted");
                this.tweets.remove(id);
            }
        }
        else
            this.tweets.put(
                id,
                new Triple<String, Float, String>("pos", pos, text));
    }
    else if (input.contains("neg_score"))
    {
        Float neg = input.getFloat(input.fieldIndex("neg_score"));
        if (this.tweets.containsKey(id))
        {
            Triple<String, Float, String> triple = this.tweets.get(id);
            if ("pos".equals(triple.getCar()))
                emit(collector, id, triple.getCaar(), neg, triple.getCdr());
            else
            {
                LOGGER.warn("one sided join attempted");
                this.tweets.remove(id);
            }
        }
        else
            this.tweets.put(
                id,
                new Triple<String, Float, String>("neg", neg, text));
    }
#+END_SRC

Where ~emit~ is defined simply by:

#+BEGIN_SRC java
    private void emit(
        BasicOutputCollector collector,
        Long id,
        String text,
        float pos,
        float neg)
    {
        collector.emit(new Values(id, pos, neg, text));
        this.tweets.remove(id);
    }
#+END_SRC

**** Deciding the Winning Class

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

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"));
    Float neg = input.getFloat(input.fieldIndex("neg_score"));
    String score = input.getString(input.fieldIndex("score"));
    String tweet_score =
        String.format("%s,%s,%f,%f,%s\n", id, tweet, pos, neg, score);
    this.tweet_scores.add(tweet_score);
    if (this.tweet_scores.size() >= 1000)
    {
        writeToHDFS();
        this.tweet_scores = new ArrayList<String>(1000);
    }
#+END_SRC

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"));
    hdfs = FileSystem.get(conf);
    file = new Path(
        Properties.getString("rts.storm.hdfs_output_file") + this.id);
    if (hdfs.exists(file))
        os = hdfs.append(file);
    else
        os = hdfs.create(file);
    wd = new BufferedWriter(new OutputStreamWriter(os, "UTF-8"));
    for (String tweet_score : tweet_scores)
    {
        wd.write(tweet_score);
    }
#+END_SRC

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"));
    Float neg = input.getFloat(input.fieldIndex("neg_score"));
    String score = input.getString(input.fieldIndex("score"));
    HttpPost post = new HttpPost(this.webserver);
    String content = String.format(
        "{\"id\": \"%d\", "  +
        "\"text\": \"%s\", " +
        "\"pos\": \"%f\", "  +
        "\"neg\": \"%f\", "  +
        "\"score\": \"%s\" }",
        id, tweet, pos, neg, score);

    try
    {
        post.setEntity(new StringEntity(content));
        HttpResponse response = client.execute(post);
        org.apache.http.util.EntityUtils.consume(response.getEntity());
    }
    catch (Exception ex)
    {
        LOGGER.error("exception thrown while attempting post", ex);
        LOGGER.trace(null, ex);
        reconnect();
    }
#+END_SRC

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
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
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
servers to handle the increase in posts, and/ or a better batching process.

** Summary

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
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

-  [[storm][Apache Storm Project Page]]

-  [[storm-multi-lingual][Storm Multi-Language Documentation]]

-  [[kafka][Apache Kafka Project Page]]

-  [[kafka-benchmark][LinkedIn Kafka Benchmarking: 2 million writes per
  second]]

-  [[storm-integrates][Storm Integration Documentation]]

-  [[supervisord][Supervisord Project Page]]

-  [[docker][Docker IO Project Page]]

-  [[storm-kafka-extern-github][Storm-Kafka Source]]

-  [[storm-sample-project][Full Source of Test Project]]

-  [[storm-incubation-proposal][Apache Storm Incubation Proposal]]

-  [[jackson-databind][Jackson Databind Project Bag]]

-  [[wiki-bag-of-words][Wikipedia: Bag of words]]

-  [[storm-trident-overview][Storm Trident API Overview]]

-  [[wiki-srp][Wikipedia: Single responsibility principle]]

-  [[wiki-stemming][Wikipedia: Stemming]]

-  [[storm-common-patterns][Storm Documentation: Common Patterns]]

-  [[storm-docs-stream-grouping][Stream Groupings]]