Kafka joins the Stream Processing club

Kafka Streams is a library to build streaming applications using Kafka topics as input/output. Kafka Streams is in the same league as other streaming systems such as: Apache Storm, Apache Flink and, not surprisingly, Apache Samza which also uses Kafka for input or output of streams.

One of the main advantages is that if you’re already using Apache Kafka and you need real-time processing, you just need to use the library and you are good to go.
Other important features are: stateful processing, windowing and ability to be deployed using your preferred solution: a simple command line, Mesos, YARN or kubernetes and docker if you’re a container party boy.

Streams and Tables

One of the key concepts in Kafka Streams is the support of KStream and KTable.
That isn’t a new concept if you come from the Event Sourcing world: the KStream is the append-only event store where its state is given by replaying the events from the beginning of time until the last event whereas KTable is the snapshot or projection of the current state of the stream given a point in time.

Example: Twitter Hashtags Job

Kafka Streams: KStream and KTable

Show me the code!

You can find the complete example here: https://github.com/mserrate/kafka-streams-app

KStream<String, JsonNode> source = builder.stream(stringDeserializer, jsonDeserializer, "streams-hashtag-input");

KTable<String, Long> counts = source
        .filter(new HashtagFilter())
        .flatMapValues(new HashtagSplitter())
        .map(new HashtagMapper())
        .countByKey(stringSerializer, longSerializer, stringDeserializer, longDeserializer, "Counts");

counts.to("streams-hashtag-count-output", stringSerializer, longSerializer);

For this example I’ve been using a simple TweetProducer who connects to the Twitter Streaming API and sends JSON events to a Kafka topic.
This topic is read as a KStream and then we begin the process:

1. Filter out the tweets without hashtags
2. Apply a flatMapValues (we are just interested in the values, not the keys) to split the different hashtags in a tweet
3. Apply a map to return a key (hashtag) value (hashtag) as we want to aggregate by hashtag
4. Aggregate the streams per key (the hashtag) and count them

Finally we send the KTable to the output queue.

Twitter Topology

The declaration of the storm topology using KafkaSpout to read the tweets from a kafka queue:

public class TwitterProcessorTopology extends BaseTopology {

    public TwitterProcessorTopology(String configFileLocation) throws Exception {
        super(configFileLocation);
    }

    private void configureKafkaSpout(TopologyBuilder topology) {
        BrokerHosts hosts = new ZkHosts(topologyConfig.getProperty("zookeeper.host"));

        SpoutConfig spoutConfig = new SpoutConfig(
                hosts,
                topologyConfig.getProperty("kafka.twitter.raw.topic"),
                topologyConfig.getProperty("kafka.zkRoot"),
                topologyConfig.getProperty("kafka.consumer.group"));
        spoutConfig.scheme= new SchemeAsMultiScheme(new StringScheme());

        KafkaSpout kafkaSpout= new KafkaSpout(spoutConfig);
        topology.setSpout("twitterSpout", kafkaSpout);
    }

    private void configureBolts(TopologyBuilder topology) {
        // filtering
        topology.setBolt("twitterFilter", new TwitterFilterBolt(), 4)
                .shuffleGrouping("twitterSpout");

        // sanitization
        topology.setBolt("textSanitization", new TextSanitizationBolt(), 4)
                .shuffleGrouping("twitterFilter");

        // sentiment analysis
        topology.setBolt("sentimentAnalysis", new SentimentAnalysisBolt(), 4)
                .shuffleGrouping("textSanitization");

        // persist tweets with analysis to Cassandra
        topology.setBolt("sentimentAnalysisToCassandra", new SentimentAnalysisToCassandraBolt(topologyConfig), 4)
                .shuffleGrouping("sentimentAnalysis");

        // divide sentiment by hashtag
        topology.setBolt("hashtagSplitter", new HashtagSplitterBolt(), 4)
                .shuffleGrouping("textSanitization");

        // persist hashtags to Cassandra
        topology.setBolt("hashtagCounter", new HashtagCounterBolt(), 4)
                .fieldsGrouping("hashtagSplitter", new Fields("tweet_hashtag"));

        topology.setBolt("topHashtag", new TopHashtagBolt())
                .globalGrouping("hashtagCounter");

        topology.setBolt("topHashtagToCassandra", new TopHashtagToCassandraBolt(topologyConfig), 4)
                .shuffleGrouping("topHashtag");
    }

    private void buildAndSubmit() throws Exception {
        TopologyBuilder builder = new TopologyBuilder();
        configureKafkaSpout(builder);
        configureBolts(builder);

        Config config = new Config();

        //set producer properties
        Properties props = new Properties();
        props.put("metadata.broker.list", topologyConfig.getProperty("kafka.broker.list"));
        props.put("request.required.acks", "1");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        config.put(KafkaBolt.KAFKA_BROKER_PROPERTIES, props);

        StormSubmitter.submitTopology("twitter-processor", config, builder.createTopology());
    }

    public static void main(String[] args) throws Exception {
        String configFileLocation = args[0];

        TwitterProcessorTopology topology = new TwitterProcessorTopology(configFileLocation);
        topology.buildAndSubmit();
    }
}

1. Filter Bolt

First of all, we are going to filter the tweets that we are interested in. As we are going to perform the sentiment analysis just to tweets in english, we are filtering on this property:

 
public class TwitterFilterBolt extends BaseBasicBolt {
    private static final Logger LOG = LoggerFactory.getLogger(TwitterFilterBolt.class);

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        try {
            JSONObject object = (JSONObject)JSONValue.parseWithException(tuple.getString(0));

            if (object.containsKey("lang") && "en".equals(object.get("lang"))) {
                long id = (long)object.get("id");
                String text = (String)object.get("text");
                String createdAt = (String)object.get("created_at");
                JSONObject entities= (JSONObject)object.get("entities");
                JSONArray hashtags =(JSONArray)entities.get("hashtags");
                HashSet<String> hashtagList = new HashSet<String>();
                for(Object hashtag : hashtags)
                {
                    hashtagList.add(((String)((JSONObject)hashtag).get("text")).toLowerCase());
                }

                collector.emit(new Values(id, text, hashtagList, createdAt));
            }
            else {
                LOG.debug("Ignoring non-english tweets");
            }

        } catch (ParseException e) {
            LOG.error("Error parsing tweet: " + e.getMessage());
        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("tweet_id", "tweet_text", "tweet_hashtags", "tweet_created_at"));
    }
}

2. Sanitization Bolt

Then, we will sanitise our tweets by converting accented characters into unaccented characters and by removing single letters or numbers:

public class TextSanitizationBolt extends BaseBasicBolt {
    private static final Logger LOG = LoggerFactory.getLogger(TextSanitizationBolt.class);

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {

        String text = tuple.getString(1);
        String normalizedText = Normalizer.normalize(text, Normalizer.Form.NFD);
        text = normalizedText.replaceAll("\\p{InCombiningDiacriticalMarks}+", "");
        text = text.replaceAll("[^\\p{L}\\p{Nd}]+", " ").toLowerCase();

        collector.emit(new Values(
                tuple.getLongByField("tweet_id"),
                text,
                tuple.getValueByField("tweet_hashtags"),
                tuple.getStringByField("tweet_created_at")));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("tweet_id", "tweet_text", "tweet_hashtags", "tweet_created_at"));
    }
}

3a. Sentiment Analysis Bolt

On this bolt we are scoring the tweet by each of its words using SentiWordNet. That’s not the best way to do it as it can have false positives or negatives given that it does the classification word by word independently: it does not cover the tweet context or sarcasm, etc. but that’s ok for a sample

public class SentimentAnalysisBolt extends BaseBasicBolt {
    private static final Logger LOG = LoggerFactory.getLogger(SentimentAnalysisBolt.class);
    SentiWordNet sentiWordNet;

    @Override
    public void prepare(Map stormConf, TopologyContext context) {
        try {
            sentiWordNet = SentiWordNet.getInstance();
        } catch (IOException e) {
            LOG.error("Problem parsing SentiWordNet file: " + e.getMessage());
        }
    }

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        double count = 0;
        String text = tuple.getStringByField("tweet_text");

        try {
            String delimiters = "\\W";
            String[] tokens = text.split(delimiters);
            double feeling = 0;
            for (int i = 0; i < tokens.length; ++i) {
                if (!tokens[i].isEmpty()) {
                    // Search as adjective
                    feeling = sentiWordNet.extract(tokens[i], "a");
                    count += feeling;
                }
            }

            LOG.info("text: " + text + " count: " + count);
        }
        catch (Exception e) {
            LOG.error("Problem found when classifying the text: " + e.getMessage());
        }

        collector.emit(new Values(
                tuple.getLongByField("tweet_id"),
                text,
                count,
                tuple.getValueByField("tweet_hashtags"),
                tuple.getStringByField("tweet_created_at")));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("tweet_id", "tweet_text", "tweet_sentiment", "tweet_hashtags", "tweet_created_at"));
    }
}

4a. Save results to Cassandra

Finally, we will store to Cassandra the the tweet with the score and its hashtags if any:

public class SentimentAnalysisToCassandraBolt extends CassandraBaseBolt {
    private static final Logger LOG = LoggerFactory.getLogger(SentimentAnalysisToCassandraBolt.class);

    public SentimentAnalysisToCassandraBolt(Properties properties) {
        super(properties);
    }

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {

        HashSet<String> hashtags = (HashSet<String>)tuple.getValueByField("tweet_hashtags");

        Statement statement = QueryBuilder.update("tweet_sentiment_analysis")
                .with(QueryBuilder.set("tweet", tuple.getStringByField("tweet_text")))
                .and(QueryBuilder.set("sentiment", tuple.getDoubleByField("tweet_sentiment")))
                .and(QueryBuilder.addAll("hashtags", hashtags))
                .and(QueryBuilder.set("created_at", tuple.getStringByField("tweet_created_at")))
                .where(QueryBuilder.eq("tweet_id", tuple.getLongByField("tweet_id")));

        LOG.debug(statement.toString());

        session.execute(statement);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }
}

3b. Hashtag Splitter Bolt

That branch of the topology graph is responsible for splitting the different hashtags and emitting a tuple per hashtag to the next bolt. That’s why we inherit from BaseRichBolt in order to manually ACK the tuple after all hashtags have been emitted.

public class HashtagSplitterBolt extends BaseRichBolt {
    OutputCollector collector;
    Map<String, Integer> count = new HashMap<String, Integer>();

    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple tuple) {
        HashSet<String> hashtags = (HashSet<String>)tuple.getValueByField("tweet_hashtags");

        for (String hashtag : hashtags) {
            collector.emit(new Values(hashtag));
        }

        collector.ack(tuple);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("tweet_hashtag"));
    }
}

4b. Hashtag Counter Bolt

In this particular case, we are using Fields Grouping because we want to partition the stream by hashtag. It means that the same hashtag will always go to the same task. Thus, we can use a hashmap to count the number of occurrences of a hashtag:

public class HashtagCounterBolt extends BaseBasicBolt {
    private static final Logger LOG = LoggerFactory.getLogger(HashtagCounterBolt.class);
    private Map<String, Long> hashtag_count = new HashMap<String, Long>();

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        String hashtag = tuple.getStringByField("tweet_hashtag");
        Long count = hashtag_count.get(hashtag);

        if (count == null)
            count = 0L;

        count++;
        hashtag_count.put(hashtag, count);

        collector.emit(new Values(hashtag, count));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("hashtag", "count"));
    }
}

5b.  Top Hashtag Bolt

For that bolt, we are using Global Grouping to get the top 20 hashtags. Global grouping means that all hashtags will go to the same bolt’s task. For that use case, we need a sliding windows in order to get the top 20 hashtags every 10 seconds. We are relying on the Storm Tick Tuple feature. For normal tuples we just do the ranking of hashtags and, when a tick tuple is received (configured to get it every 10sec) we emit the ranking calculated over this window of time.

public class TopHashtagBolt extends BaseBasicBolt {
    List<List> rankings = new ArrayList<List>();
    private static final Logger LOG = LoggerFactory.getLogger(TopHashtagBolt.class);
    private static final Integer TOPN = 20;
    private static final Integer TICK_FREQUENCY = 10;

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        if (isTickTuple(tuple)) {
            LOG.debug("Tick: " + rankings);
            collector.emit(new Values(new ArrayList(rankings)));
        } else {
            rankHashtag(tuple);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("tophashtags"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        Config conf = new Config();
        conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, TICK_FREQUENCY);
        return conf;
    }

    private void rankHashtag(Tuple tuple) {
        String hashtag = tuple.getStringByField("hashtag");
        Integer existingIndex = find(hashtag);
        if (null != existingIndex)
            rankings.set(existingIndex, tuple.getValues());
        else
            rankings.add(tuple.getValues());

        Collections.sort(rankings, new Comparator<List>() {
            @Override
            public int compare(List o1, List o2) {
                return compareRanking(o1, o2);
            }
        });

        shrinkRanking();
    }

    private Integer find(String hashtag) {
        for(int i = 0; i < rankings.size(); ++i) { String current = (String) rankings.get(i).get(0); if (current.equals(hashtag)) { return i; } } return null; } private int compareRanking(List one, List two) { long valueOne = (Long) one.get(1); long valueTwo = (Long) two.get(1); long delta = valueTwo - valueOne; if(delta > 0) {
            return 1;
        } else if (delta < 0) { return -1; } else { return 0; } } private void shrinkRanking() { int size = rankings.size(); if (TOPN >= size) return;
        for (int i = TOPN; i < size; i++) {
            rankings.remove(rankings.size() - 1);
        }
    }

    private static boolean isTickTuple(Tuple tuple) {
        return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID)
                && tuple.getSourceStreamId().equals(Constants.SYSTEM_TICK_STREAM_ID);
    }
}

6b. Save results to Cassandra

Finally, we are storing the top N hashtags per day in Cassandra. For that we’re using the row partitioning pattern to store a row per day and the top hashtags for each time bucket (20 seconds)

public class TopHashtagToCassandraBolt extends CassandraBaseBolt {
    private static final Logger LOG = LoggerFactory.getLogger(TopHashtagToCassandraBolt.class);

    public TopHashtagToCassandraBolt(Properties properties) {
        super(properties);
    }

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        List<List> rankings = (List) tuple.getValue(0);

        Map<String, Long> rankingMap = new HashMap<>();

        for (List list : rankings) {
            rankingMap.put((String) list.get(0), (Long) list.get(1));
        }

        DateFormat df = new SimpleDateFormat("yyyy-MM-dd");

        Statement statement = QueryBuilder.insertInto("top_hashtag_by_day")
                .value("date", df.format(new Date()))
                .value("bucket_time", QueryBuilder.raw("dateof(now())"))
                .value("ranking", rankingMap);

        LOG.debug(statement.toString());

        session.execute(statement);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}

Next steps

Although the hashtag counter may work, I will not say that is entirely correct and there are better ways to do it:

• You can take a look to this excellent resource: http://www.michael-noll.com/blog/2013/01/18/implementing-real-time-trending-topics-in-storm/
• Using Storm Trident: In the next post I will show how to use the high level abstraction from Storm that allows to process a stream as a sequence of small batches of data (aka micro-batching) and fits better for the top hashtags example.

Overview

The solution consists of the following:

• twitter-kafka-producer: A very basic producer that reads tweets from the Twitter Streaming API and stores them in Kafka.
• twitter-storm-topology: A Storm topology that reads tweets from Kafka and, after applying filtering and sanitization, process the messages in parallel for:
  • Sentiment Analysis: Using a sentiment analysis algorithm to classify the tweet into a positive or negative feeling.
  • Top Hashtags: Calculates the top 20 hashtags using a sliding window.

Storm Topology

The Storm topology consist of the following elements:

• Kafka Spout: The spout implementation to read messages from Kafka.
• Filtering: Filtering out all non-english language tweets.
• Sanitization: Text normalization in order to be processed properly by the sentiment analysis algorithm.
• Sentiment Analysis: The algorithm that analyses word by word the text of the tweet, giving a value between -1 to 1.
• Sentiment Analysis to Cassandra: Stores the tweets and its sentiment value in Cassandra.
• Hashtag Splitter: Splits the different hashtags appearing in a tweet.
• Hashtag Counter: Counts hashtag occurrences.
• Top Hashtag: Does a ranking of the top 20 hashtags given a sliding windows (using the Tick Tuple feature from Storm).
• Top Hashtag to Cassandra: Stores the top 20 hashtags in Cassandra.

Summary

In this post we have seen the benefits of using Apache Kafka & Apache Storm to ingest and process streams of data, on next posts will look at the implementation details and will provide some analytical insight from the data stored in Cassandra.

The sample can be found on Github: https://github.com/mserrate/twitter-streaming-app,

Big Data

[note color="#ebebeb"]Current View = Query(Batch View) + Query(Stream View)[/note]

Batch Layer

Stream Layer

You can see the details in the following link:

The IWantToRunBeforeConfiguration interface is a great help in order to embrace the DRY principle.

Just define your implementation in an assembly referenced by all the endpoints:

public class UnobtrusiveConventions : IWantToRunBeforeConfiguration
{
    public void Init()
    {
        Configure.Instance
            .DefiningCommandsAs(t => t.Namespace != null
                && t.Namespace.EndsWith("Commands"))
            .DefiningEventsAs(t => t.Namespace != null
                && t.Namespace.EndsWith("Events"))
            .DefiningMessagesAs(t => t.Namespace != null
                && t.Namespace.EndsWith("Messages"));
    }
}

and NServiceBus will pick this class automatically for each endpoint.

• November 26. Barcelona
• December 10. Madrid

More info at: http://udidahan.com/2012/10/12/training-for-this-winter/

Cassandra installation

Composition with ASP.NET MVC

1. Install RazorGenerator from VS gallery
2. On each class library, install RazorGenerator.Mvc from Nuget
3. Create the folder structure following MVC conventions (Controllers, Views, etc.)
4. Add a web.config to get intellisense for your views
5. Change "Custom Tool" to RazorGenerator on your views in order to precompile it