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

问题导读
1.编写Topology，由哪三步分组成？
2.创建maven工程的作用是什么？
3.如何编写编写Spout、bolt、topo？

本文以一个简单的example来讲解如何开发storm应用程序

1、创建maven工程

　　在eclipse下创建maven工程

　　

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

其实我想问的是，这个storm集群或者单机如果是在linux中配置，在Windows中的eclipse上开发的话。

main方法里面总要配置一点什么吧，比如集群nimbus的IP地址等，不然一个windows上的程序怎么会知道storm集群的位置呢。难道windows中也需要安装配置storm吗，具体怎么配置呢？

梦回三国 发表于 2014-9-29 16:58
其实我想问的是，这个storm集群或者单机如果是在linux中配置，在Windows中的eclipse上开发的话。

main方 ...

https://github.com/nathanmarz/storm/blob/master/conf/defaults.yaml。

~/.storm/storm.yaml
修改storm.yaml的默认配置。改一下：
1  #storm.local.dir: "/mnt/storm"，改成适合自己平台的地址。这文件夹存储storm的本地文件。
2  #storm.cluster.mode: "distributed" ，采取的远程模式，而不是本地模式。
3  #nimbus.host:”10.141.9.227”,nimbus的地址。

howtodown 发表于 2014-9-29 20:02
https://github.com/nathanmarz/storm/blob/master/conf/defaults.yaml。

~/.storm/storm.yaml

谢谢，我试一下

howtodown 发表于 2014-9-29 20:02
https://github.com/nathanmarz/storm/blob/master/conf/defaults.yaml。

~/.storm/storm.yaml

#storm.cluster.mode: "distributed" 是本地平台的storm改呢，还是linux集群中的改呢，还是都要修改？

梦回三国 发表于 2014-9-29 22:16
#storm.cluster.mode: "distributed" 是本地平台的storm改呢，还是linux集群中的改呢，还是都要修改？

这个只是针对的，开发。一般只需要该


nimbus.host:”10.141.9.227”,
nimbus的地址。
其它的可以先不用改。

楼主
请教个本地模式问题
如果创建java工程后（不是maven工程），直接用eclipse run as java application的话，报错，报java.lang.RuntimeException: Error when launching multilang subprocess
和Caused by: java.io.IOException: Cannot run program "python" (in directory "C:\Users\QZJ\AppData\Local\Temp\3c2a52cd-2e02-491f-b4a7-240bd4ac7e17\supervisor\stormdist\word-count-1-1427536552\resources"): CreateProcess error=2, ?????????
是什么原因？
如果不通过run as 运行，而是把工程导出成jar包，然后在windows cmd控制台运行
storm jar starter.jar storm.starter.WordCountTopology则能正常运行，控制台输出调试信息，这是什么原因？
如果创建的是maven工程，则通过run as 直接在eclipse运行没问题（好像是这样，不是记得特别清楚了，之前我试过几次）

创建java project按照这个地址创建
http://www.cnblogs.com/XjChenny/p/3214039.html
没有创建meven工程，而是创建的java project只是创建完以后，只能通过 windows cmd storm命令提交jar，如果在wordCountTopology类里点击右键运行run as java application的话，本地模式运行，则报错；
报错的原因跟有没有装python没关系，跟resources也没有关系，我装不装python都能通过storm提交

qzjqzjqzj 发表于 2015-3-28 18:33
创建java project按照这个地址创建
http://www.cnblogs.com/XjChenny/p/3214039.html
没有创建meven工程 ...

参考这个：
storm本地模式是不是本地程序直接可以跑起来？

谢谢楼主分享。