I'm trying to implement a Kafka Spout inside a Storm Topology.
After looking at various tutorials this is the code I finished with: import java.util.UUID;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
//import storm configuration packages
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.ZkHosts;
import backtype.storm.spout.SchemeAsMultiScheme;
//Create main class TestTopo submit topology.
public class TestTopo {
public static void main(String[] args) throws Exception{
String zkHost = "sandbox.hortonworks.com:2181";
//Create Config instance for cluster configuration
Config config = new Config();
config.setDebug(true);
TopologyBuilder builder = new TopologyBuilder();
SpoutConfig MyKafkaConfig = new SpoutConfig(new ZkHosts(zkHost), "topic", "", UUID.randomUUID().toString());
builder.setSpout("MyKafkaSpout", new KafkaSpout(MyKafkaConfig));
builder.setBolt("consoleOutputBolt", new ConsoleBolt())
.shuffleGrouping("MyKafkaSpout");
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("TestTopo", config, builder.createTopology());
Thread.sleep(10000);
//Stop the topology
cluster.shutdown();
}
}
The main problem I have is that there is a dependency problem with the storm.kafka library which looks like is not available with the main storm functionality. It looks like it is not included in the Sandbox. Where I can find this implementation?? Thanks!
... View more
