5

我正在使用 KafkaSpout。请在下面找到测试程序。

我正在使用 Storm 0.8.1。Storm 0.8.2 中有 Multischeme 类。我会用那个。我只想通过实例化 StringScheme() 类来了解早期版本是如何工作的?我在哪里可以下载早期版本的 Kafka Spout?但我怀疑这不是在 Storm 0.8.2 上工作的正确选择。???(使困惑)

当我在storm集群上运行代码(如下所示)时(即当我推送我的拓扑时)我得到以下错误(当Scheme部分被注释时会发生这种情况,当然我会得到编译器错误,因为该类在0.8中不存在.1):

java.lang.NoClassDefFoundError: backtype/storm/spout/MultiScheme
        at storm.kafka.TestTopology.main(TestTopology.java:37)
Caused by: java.lang.ClassNotFoundException: backtype.storm.spout.MultiScheme

在下面给出的代码中,您可能会发现 spoutConfig.scheme=new StringScheme(); 部分评论。如果我不评论那条很自然的行,我会收到编译器错误,因为那里没有构造函数。此外,当我实例化 MultiScheme 时,我会收到错误,因为我在 0.8.1 中没有该类。

public class TestTopology {
    public static class PrinterBolt extends BaseBasicBolt {
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
        }

        public void execute(Tuple tuple, BasicOutputCollector collector) {
            System.out.println(tuple.toString());
        }
    }

    public static void main(String [] args) throws Exception {
        List<HostPort> hosts = new ArrayList<HostPort>();
        hosts.add(new HostPort("127.0.0.1",9092));
        LocalCluster cluster = new LocalCluster();
        TopologyBuilder builder = new TopologyBuilder();
        SpoutConfig spoutConfig = new SpoutConfig(new KafkaConfig.StaticHosts(hosts, 1), "test", "/zkRootStorm", "STORM-ID");
        spoutConfig.zkServers=ImmutableList.of("localhost");
        spoutConfig.zkPort=2181;
        //spoutConfig.scheme=new StringScheme();
        spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
        builder.setSpout("spout",new KafkaSpout(spoutConfig));
        builder.setBolt("printer", new PrinterBolt())
                .shuffleGrouping("spout");
        Config config = new Config();

        cluster.submitTopology("kafka-test", config, builder.createTopology());

        Thread.sleep(600000);
    }
4

2 回答 2

9

我有同样的问题。终于解决了,我把完整的运行示例放到了github上。

欢迎您在这里查看> https://github.com/buildlackey/cep

(单击storm+kafka 目录可以找到一个可以帮助你启动并运行的示例程序)。

于 2013-10-25T21:59:04.673 回答
5

我们有一个类似的问题。

我们的解决方案:

  1. 打开 pom.xml

  2. 将范围从提供更改为<scope>compile</scope>

如果您想了解有关依赖范围的更多信息,请查看 Maven 文档: Maven 文档 - 依赖范围

于 2014-10-09T13:43:11.453 回答