原创文章,转载请注明: 转载自http://www.cnblogs.com/tovin/p/3971113.html
本文以一个简单的example来讲解如何开发storm应用程序
1、创建maven工程
在eclipse下创建maven工程,可以参照http://www.cnblogs.com/tovin/p/3822985.html
2、修改pom.xm添加依赖包
使用maven-assembly-plugin插件将工程依赖的jar都一起打包
storm的<scope>设置provided,主要是因为只要编译时需要storm包,当在storm集群运行时就不要将它一起打包了。
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.test</groupId> <artifactId>storm-example</artifactId> <version>0.0.1-SNAPSHOT</version> <packaging>jar</packaging> <name>storm-example</name> <url>http://maven.apache.org</url> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties> <dependencies> <dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-core</artifactId> <version>0.9.2-incubating</version> <scope>provided</scope> </dependency> </dependencies> <build> <plugins> <plugin> <artifactId>maven-assembly-plugin</artifactId> <version>2.4</version> <configuration> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> </configuration> <executions> <execution> <id>make-assembly</id> <phase>package</phase> <goals> <goal>single</goal> </goals> </execution> </executions> </plugin> </plugins> </build> </project>
3、编写Topology
(1) 编写Spout
import backtype.storm.spout.SpoutOutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.base.BaseRichSpout; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Values; public class RandomSpout extends BaseRichSpout{ private SpoutOutputCollector collector; private static String[] words = {"happy","excited","angry"}; /* (non-Javadoc) * @see backtype.storm.spout.ISpout#open(java.util.Map, backtype.storm.task.TopologyContext, backtype.storm.spout.SpoutOutputCollector) */ public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector arg2) { // TODO Auto-generated method stub this.collector = arg2; } /* (non-Javadoc) * @see backtype.storm.spout.ISpout#nextTuple() */ public void nextTuple() { // TODO Auto-generated method stub String word = words[new Random().nextInt(words.length)]; collector.emit(new Values(word)); } /* (non-Javadoc) * @see backtype.storm.topology.IComponent#declareOutputFields(backtype.storm.topology.OutputFieldsDeclarer) */ public void declareOutputFields(OutputFieldsDeclarer arg0) { // TODO Auto-generated method stub arg0.declare(new Fields("randomstring")); } }
(2)编写bolt
import backtype.storm.topology.BasicOutputCollector; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.base.BaseBasicBolt; import backtype.storm.tuple.Tuple; public class SenqueceBolt extends BaseBasicBolt{ /* (non-Javadoc) * @see backtype.storm.topology.IBasicBolt#execute(backtype.storm.tuple.Tuple, backtype.storm.topology.BasicOutputCollector) */ public void execute(Tuple input, BasicOutputCollector collector) { // TODO Auto-generated method stub String word = (String) input.getValue(0); String out = "I‘m " + word + "!"; System.out.println("out=" + out); } /* (non-Javadoc) * @see backtype.storm.topology.IComponent#declareOutputFields(backtype.storm.topology.OutputFieldsDeclarer) */ public void declareOutputFields(OutputFieldsDeclarer declarer) { // TODO Auto-generated method stub } }
(3)编写topo
提供cluster和Local两种运行模式,这样我们就很方便的在本地运行FirstTopo来调试我们程序了。
import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.StormSubmitter; import backtype.storm.topology.TopologyBuilder; import backtype.storm.utils.Utils; public class FirstTopo { public static void main(String[] args) throws Exception { TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("spout", new RandomSpout()); builder.setBolt("bolt", new SenqueceBolt()).shuffleGrouping("spout"); Config conf = new Config(); conf.setDebug(false); if (args != null && args.length > 0) { conf.setNumWorkers(3); StormSubmitter.submitTopology(args[0], conf, builder.createTopology()); } else { LocalCluster cluster = new LocalCluster(); cluster.submitTopology("firstTopo", conf, builder.createTopology()); Utils.sleep(100000); cluster.killTopology("firstTopo"); cluster.shutdown(); } } }
(4)运行结果
本地运行时,在eclipse中的输出:
在storm集群中运行时输出可以通过Storm UI进行查看
原创文章,转载请注明: 转载自http://www.cnblogs.com/tovin/p/3971113.html