Apache Storm工作实例

eption{ //Create Config instance for cluster configuration Config config = new Config(); config.setDebug(true); // TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("call-log-reader-spout", new FakeCallLogReaderSpout()); builder.setBolt("call-log-creator-bolt", new CallLogCreatorBolt()) .shuffleGrouping("call-log-reader-spout"); builder.setBolt("call-log-counter-bolt", new CallLogCounterBolt()) .fieldsGrouping("call-log-creator-bolt", new Fields("call")); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("LogAnalyserStorm", config, builder.createTopology()); Thread.sleep(10000); //Stop the topology cluster.shutdown(); }}构建和运行应用程序完整的应用程序有四个Java代码。它们是 - FakeCallLogReaderSpout.java CallLogCreaterBolt.java CallLogCounterBolt.java LogAnalyerStorm.java 应用程序可以使用以下命令构建 - javac -cp “/path/to/storm/apache-storm-0.9.5/lib/*” *.java应用程序可以使用以下命令运行 - java -cp “/path/to/storm/apache-storm-0.9.5/lib/*”:. LogAnalyserStorm输出一旦应用程序启动，它将输出有关集群启动过程，spout和螺栓处理的完整详细信息，最后是集群关闭过程。在“CallLogCounterBolt”中，我们打印了呼叫及其计数详细信息。此信息将显示在控制台上如下 - 1234123402 - 1234123401 : 781234123402 - 1234123404 : 881234123402 - 1234123403 : 1051234123401 - 1234123404 : 741234123401 - 1234123403 : 811234123401 - 1234123402 : 811234123403 - 1234123404 : 861234123404 - 1234123401 : 631234123404 - 1234123402 : 821234123403 - 1234123402 : 831234123404 - 1234123403 : 861234123403 - 1234123401 : 93非JVM语言Storm拓扑通过Thrift接口实现，这使得轻松地提交任何语言的拓扑。Storm支持Ruby，Python和许多其他语言。让我们来看看python绑定。 Python绑定 Python是一种通用的解释，交互，面向对象和高级编程语言。Storm支持Python实现其拓扑。Python支持发射，锚定，acking和日志操作。如你所知，bolt可以用任何语言定义。用另一种语言编写的bolt作为子进程执行，Storm通过stdin / stdout与JSON消息进行通信。首先拿一个支持python绑定的样例bolt WordCount。public static class WordCount implements IRichBolt { public WordSplit() { super("python", "splitword.py"); } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word")); }}这里的类WordCount实现IRichBolt接口和运行与python实现指定超级方法参数“splitword.py”。现在创建一个名为“splitword.py”的python实现。import storm class WordCountBolt(storm.BasicBolt): def process(self, tup): words = tup.values[0].split(" ") for word in words: storm.emit([word])WordCountBolt().run()这是Python的示例实现，它计算给定句子中的单词。同样，您也可以与其他支持语言绑定。