Storm应用最基本的例子
1. 建立Maven项目
我们用Maven来管理项目,方便lib依赖的引用和版本控制。
建立最基本的pom.xml如下:
- <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.edi.storm</groupId>
- <artifactId>storm-samples</artifactId>
- <version>0.0.1-SNAPSHOT</version>
- <packaging>jar</packaging>
- <properties>
- <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
- </properties>
- <repositories>
- <repository>
- <id>clojars.org</id>
- <url>http://clojars.org/repo</url>
- </repository>
- </repositories>
- <build>
- <finalName>storm-samples</finalName>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-compiler-plugin</artifactId>
- <version>3.1</version>
- <configuration>
- <source>1.7</source>
- <target>1.7</target>
- <encoding>${project.build.sourceEncoding}</encoding>
- </configuration>
- </plugin>
- <plugin>
- <artifactId>maven-assembly-plugin</artifactId>
- <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>
- <dependencies>
- <dependency>
- <groupId>storm</groupId>
- <artifactId>storm</artifactId>
- <version>0.9.0-rc2</version>
- <scope>provided</scope>
- </dependency>
- </dependencies>
- </project>
maven-compiler-plugin : 为了方便指定编译时jdk。Storm的依赖包里面某些是jdk1.5的.
和
maven-assembly-plugin: 为了把所有依赖包最后打到一个jar包去,方便测试和部署。后面会提到如果不想打到一个jar该怎么做。
2. 建立Spout
前文提到过,Storm中的spout负责发射数据。我们来实现这样一个spout:
它会随机发射一系列的句子,句子的格式是 谁:说的话
代码如下:
- public class RandomSpout extends BaseRichSpout {
- private SpoutOutputCollector collector;
- private Random rand;
- private static String[] sentences = new String[] {"edi:I'm happy", "marry:I'm angry", "john:I'm sad", "ted:I'm excited", "laden:I'm dangerous"};
- @Override
- public void open(Map conf, TopologyContext context,
- SpoutOutputCollector collector) {
- this.collector = collector;
- this.rand = new Random();
- }
- @Override
- public void nextTuple() {
- String toSay = sentences[rand.nextInt(sentences.length)];
- this.collector.emit(new Values(toSay));
- }
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields("sentence"));
- }
- }
这里要先理解Tuple的概念。
Storm中,基本元数据是靠Tuple才承载的。或者说,Tuple是数据的一个大抽象。它要求实现类必须能序列化。
a. 用collector.emit()方法发射tuple。我们不用自己实现tuple,我们只需要定义tuple的value,Storm会帮我们生成tuple。Values对象接受变长参数。Tuple中以List存放Values,List的Index按照new Values(obj1, obj2,...)的参数的index,例如我们emit(new Values("v1", "v2")), 那么Tuple的属性即为:{ [ "v1" ], [ "V2" ] }
b. declarer.declare方法用来给我们发射的value在整个Stream中定义一个别名。可以理解为key。该值必须在整个topology定义中唯一。
3. 建立Bolt
既然有了源,那么我们就来建立节点处理源流出来的数据。怎么处理呢?为了演示,我们来做些无聊的事情:末尾添加"!",然后打印。
两个功能,两个Bolt。
先看添加"!"的Bolt
- public class ExclaimBasicBolt extends BaseBasicBolt {
- @Override
- public void execute(Tuple tuple, BasicOutputCollector collector) {
- //String sentence = tuple.getString(0);
- String sentence = (String) tuple.getValue(0);
- String out = sentence + "!";
- collector.emit(new Values(out));
- }
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields("excl_sentence"));
- }
- }
在RandomSpout中,我们发射的Tuple具有这样的属性 { [ "edi:I'm Happy" ] }, 所以tuple的value list中第0个值,肯定是个String。我们用tuple.getvalue(0)取到。
Storm为tuple封装了一些方法方便我们取一些基本类型,例如String,我们可以直接用getString(int N) 。
取到以后,我们在末尾添加"!"后,仍然发射一个Tuple,定义其唯一的value的field 名字为"excl_sentence"
打印Bolt
- public class PrintBolt extends BaseBasicBolt {
- @Override
- public void execute(Tuple tuple, BasicOutputCollector collector) {
- String rec = tuple.getString(0);
- System.err.println("String recieved: " + rec);
- }
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- // do nothing
- }
- }
仍然是取第一个,因为我们并没有定义过第二个value
4. 建立Topology
现在我们建立拓扑结构的主要组件都有了,可以创建topology了。
- public class ExclaimBasicTopo {
- public static void main(String[] args) throws Exception {
- TopologyBuilder builder = new TopologyBuilder();
- builder.setSpout("spout", new RandomSpout());
- builder.setBolt("exclaim", new ExclaimBasicBolt()).shuffleGrouping("spout");
- builder.setBolt("print", new PrintBolt()).shuffleGrouping("exclaim");
- 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("test", conf, builder.createTopology());
- Utils.sleep(100000);
- cluster.killTopology("test");
- cluster.shutdown();
- }
- }
- }
很简单,对吧。
其中,
- builder.setSpout("spout", new RandomSpout());
- builder.setBolt("exclaim", new ExclaimBasicBolt()).shuffleGrouping("spout");
- builder.setBolt("print", new PrintBolt()).shuffleGrouping("exclaim");
- .shuffleGrouping
可以看到,如果我们运行时不带参数,是把topology提交到了LocalCluster的,即所有的task都在一个本地JVM去执行。可以用LocalCluster来调试。如果后面带一个参数,即为该topology的名字,那么就把该topology提交到集群上去了。
把项目用M2E插件导入Eclipse直接运行试试
- String recieved: marry:I'm angry!
- String recieved: edi:I'm happy!
- String recieved: john:I'm sad!
- String recieved: edi:I'm happy!
- String recieved: ted:I'm excited!
- String recieved: laden:I'm dangerous!
- String recieved: edi:I'm happy!
- String recieved: edi:I'm happy!
这里我们并没有指定并行,那么其实是每个spout、bolt仅有一个线程对应去执行。
我们修改下代码,指定并行数
- builder.setBolt("exclaim", new ExclaimBasicBolt(), 2).shuffleGrouping("spout");
- builder.setBolt("print", new PrintBolt(),3).shuffleGrouping("exclaim");
由于我们并没有多指定task数目,所以默认,会有两个exectuor去执行两个exclaimBasicBolt的task,3个executor去执行3个PrintBolt的task。
为了方便体现确实是并行,我们修改PrintBolt代码如下:
- public class PrintBolt extends BaseBasicBolt {
- private int indexId;
- @Override
- public void prepare(Map stormConf, TopologyContext context) {
- this.indexId = context.getThisTaskIndex();
- }
- @Override
- public void execute(Tuple tuple, BasicOutputCollector collector) {
- String rec = tuple.getString(0);
- System.err.println(String.format("Bolt[%d] String recieved: %s",this.indexId, rec));
- }
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- // do nothing
- }
- }
运行下看看:
- Bolt[0] String recieved: marry:I'm angry!
- Bolt[2] String recieved: john:I'm sad!
- Bolt[2] String recieved: ted:I'm excited!
- Bolt[2] String recieved: john:I'm sad!
- Bolt[2] String recieved: john:I'm sad!
证实确实是并发了。
本地测试通过了,我们用 mvn clean install 命令编译,然后把target目录下生成的 storm-samples-jar-with-dependencies.jar 拷到nimbus机器上,执行
- ./storm jar storm-samples-jar-with-dependencies.jar com.edi.storm.topos.ExclaimBasicTopo test
看到spout 已然已经emit了 11347280个tuple了…… 而id为exclaim的bolt也已经接受了2906920个tuple了。print没有输出,所以emit为0。