CODEWITHSUNDEEP
apache-stormapache-kafka
swarup
swarup

Reputation: 56

retrieving topics from Kafka and sending to corresponding spout and bolt

I am producing multiple topic into Kafka. I want to retrieve all the topic from Kafka and I am having different spouts and bolts.And I want to send each topic to the corresponding spout and associated bolt (e.g for topic1 I am having corresponding spout1 and bolt1,for topic2 I am having corresponding spout2 and bolt2 and so on..)
How can I do this?

Upvotes: 0

Views: 3196

Answers (2)

Chris Bedford
Chris Bedford

Reputation: 2702

I created a demo kafka spout project that you should be able to use as a starting point: https://github.com/buildlackey/cep/tree/master/storm%2Bkafka.

regards /

chris

Upvotes: 3

user2720864
user2720864

Reputation: 8171

While I don't get exactly what you are trying to do (do you have separate topologies running for each topic?) normally what you can do is , in your spout1 create a consumer which will be subscribed to topic1 and emitting values as soon as it receives one. and then chain the output to corresponding bolts for further execution.

But as far as I understood you should take a look at the KafkaSpout implementation under github storm-contrib project. It is basically a spout implementation which read from the Kafka cluster and all you need is to create the configuration properly.

From the documentation it basically look like this 

    SpoutConfig spoutConfig = new SpoutConfig(
               ImmutableList.of("kafkahost1", "kafkahost2"), // list of Kafka brokers
               8, // number of partitions per host
               "clicks", // topic to read from
               "/kafkastorm", // the root path in Zookeeper for the spout to store the consumer offsets
               "discovery"); // an id for this consumer for storing the consumer offsets in Zookeeper
    KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);

One thing to mention here is the above implementation uses Kafka 0.7 so in case you are working with the latest (0.8, and you should) implementation you can find the 0.8 support here

Upvotes: 0

Related Questions