Now, let’s connect the two!
By now, we’ve assumed that you are able to get the ExclamationTopology running from our previous guide. In this guide, we are going to use Kafka as our Spout to the ExclamationTopology.
I usually check out the binary version of Storm to run Storm itself, but I like to develop on the source version of Storm. To that end, let’s grab the source version of Storm and extract it:
$ wget http://apache.mirrors.pair.com/storm/apache-storm-0.9.2-incubating/apache-storm-0.9.2-incubating-src.tar.gz
$ tar -zvxf apache-storm-0.9.2-incubating-src.tar.gz
You can see that a Kafka-connector project has already been written for us, in external/storm-kafka.
To understand how to leverage this project, let’s make Kafka as our Spout for ExclamationTopology.
First, copy the ExclamationTopology to the storm-kafka project, since that’s where all the KafkaSpout classes are (there are other ways of doing this, such as bundling storm-kakfa into a jar, and using these libraries instead).
$ cp apache-storm-0.9.2-incubating-src/examples/storm-starter/src/jvm/storm/starter/ExclamationTopology.java apache-storm-0.9.2-incubating-src/external/storm-kafka/src/jvm/storm/kafka
Now add the following imports into ExclamationTopology:
It should look something like this:
Remember also to change the package at the top of the import statements to storm.kaka (not storm.starter), as seen in the screenshot.
Then use this to setup the Kafka Spout, assuming you have created a “test” topic as shown in our Kafka guide.
String zkConnString = "ec2-54-183-196-65.us-west-1.compute.amazonaws.com";
String topicName = "test";
BrokerHosts hosts = new ZkHosts("localhost:2181");
SpoutConfig spoutConfig = new SpoutConfig(hosts, "test", "/test", "discovery");
spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
Here is a screenshot:
Then add the following lines to the pom.xml file currently in that directory. These have to go within the “plugins” bracket:
Also, remove the “scope” field in the org.apache.kafka POM dependency in the pm.xml file under storm-kaka.
Then, compile the package, and submit it to a Storm production cluster.
If you remember from our tutorial on Kafka, you can open up a Producer command line terminal and start typing messages into the Kafka. Go ahead and do this.
On the other side, you should see Storm adding exclamations to everything you type.
Congratulations, you have connected Storm with Kafka!