org.apache.storm.kafka.trident

Class TridentKafkaWordCount

java.lang.Object

org.apache.storm.kafka.trident.TridentKafkaWordCount

All Implemented Interfaces:

Serializable

public class TridentKafkaWordCount
extends Object
implements Serializable

A sample word count trident topology using transactional kafka spout that has the following components.

1. KafkaBolt that receives random sentences from RandomSentenceSpout and publishes the sentences to a kafka “test” topic.
2. TransactionalTridentKafkaSpout that consumes sentences from the “test” topic, splits it into words, aggregates and stores the word count in a MemoryMapState.
3. DRPC query that returns the word counts by querying the trident state (MemoryMapState).

For more background read the trident tutorial, trident state and Storm Kafka .

See Also:

Serialized Form

Constructor Summary

Constructors

Constructor and Description
TridentKafkaWordCount()

Method Summary

Modifier and Type	Method and Description
static void	main(String[] args) To run this topology it is required that you have a kafka broker running.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

TridentKafkaWordCount

public TridentKafkaWordCount()

Method Detail

main

public static void main(String[] args)
                 throws Exception

To run this topology it is required that you have a kafka broker running.

Create a topic ‘test’ with command line,

 kafka-topics.sh –create –zookeeper localhost:2181 –replication-factor 1 –partition 1 –topic test 

To run in local mode,

 storm jar storm-starter-topologies-{version}.jar org.apache.storm.starter.trident.TridentKafkaWordCount 

This will also run a local DRPC query and print the word counts.

To run in distributed mode, run it with a topology name. You will also need to start a drpc server and specify the drpc server details storm.yaml before submitting the topology.

 storm jar storm-starter-topologies-{version}.jar org.apache.storm.starter.trident.TridentKafkaWordCount zkhost:port broker:port wordcount 

This will submit two topologies, one for the producer and another for the consumer. You can query the results (word counts) by running an external drpc query against the drpc server.

Throws:

Exception