Windows下基于eclipse的Storm应用开发与调试

本文以一个简单的example来讲解如何开发storm应用程序
1、创建maven工程
　　在eclipse下创建maven工程，可以参照/content/4616402.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进行查看
　　　　

原创文章，转载请注明： 转载自/article/5670120.html