标签：storm

代码编写完毕，接下来就是检验代码的效果了。
Storm的代码可以在本地上跑，也可以在集群上跑。为了模拟真实的操作环境，我们当然要在集群上跑，不然前面配置那么多就白配置了。
当然，如果要在本地上跑，拓扑的代码就不一样了，这里先不说，上一篇的代码是按照集群环境来写的。接下来，我们就要验收一下前面的工作成果了。

首先在eclipse的工程中，将代码export成jar文件，步骤很简单，右键–export–jar文件。另外，你得把拓扑的函数完整名字记录下来，就这个例子而言，拓扑类是Topmain,然而完整路径则不一定，在eclipse，直接把鼠标移动到类名上，就能看到完整路径了。如下图：

主类名是stormtest.test.TopMain，这和你的工程、包有关，反正先记录下来。

接下来把3台虚拟机都跑起来。把刚才的jar包上传到第一台服务器上，即配置的主节点。
由于zookeeper我们配置的也是在主节点上，所以在主节点服务器上需要开启zookeeper的服务。切换到zookeeper的主目录下的bin目录，然后运行以下命令

./zkServer.sh start

如上图即运行成功。

然后在这台主机上运行storm的主节点nimbus：将目录切换到storm目录下的bin，运行以下命令：

./storm nimbus 1>/dev/null 2>&1 &

没出错的话，nimbus就会在后台运行了。然后再开启UI，方便我们在线查看工程运行的情况。同样在该目录下，运行以下命令：

./storm ui 1>/dev/null 2>&1 &

输入jps可以查看进程的运行情况：

接下来，可以在本地访问UI了，在浏览器输入–主机地址:8080 回车访问。比如我就输入
192.168.254.100:8080,可以看到以下概况：

OK，接下来开启从节点。切换到第二台机器的命令行，同样切换目录到storm下的bin，输入以下命令：

./storm supervisor 1>/dev/null 2>&1 &

如法炮制，切换到第三台机器，同样运行

./storm supervisor 1>/dev/null 2>&1 &

这样，我们的集群就先运行起来了。这时候回去UI刷新一下，可以看到两个从节点确实跑起来了：

此时此刻，我们要进行最激动人心的一步了，提交拓扑。还记得刚才的jar包吗，我是把它上传到第一台机器的上了 完整路径是/home/filetrans/storm-example.jar
在第一台机器上切换目录到storm下的bin，运行以下命令：

./storm jar /home/filetrans/storm-example.jar stormtest.test.TopMain

以上命令，jar表示运行jar程序，/home/filetrans/storm-example.jar表示jar程序的路径，stormtest.test.TopMain表示主函数的完整路径，这在文章一开始有提到过。
出现以下提示，说明拓扑提交成功了。

此时再去UI刷新看看，我们会发现有拓扑在工作了，拓扑名为example_top，状态为ACTIVE：

Num Workers是我们代码里设置的worker数，有3个。Num executor为线程数，有14个，14是怎么来的呢？首先以下三行代码，我们可以看出一共有加起来就一共有12个线程了。

//设置spout，并行度为4，第一个参数为自定义的名字
topologyBuilder.setSpout("randomspout", new RandomSpout(),4);
//设置bolt，数据来源是上面的spout
topologyBuilder.setBolt("upperbolt", new UpperBolt(),4).shuffleGrouping("randomspout");
topologyBuilder.setBolt("suffixbolt", new SuffixBolt(), 4).shuffleGrouping("upperbolt");

然后以下这一行代码，有两个线程：

config.setNumAckers(2);   //应答器数目

全部加起来，就有14个线程了。

Ok，程序也跑了有一段时间了，我们去看看效果。回头看看SuffixBolt类
我们在里面写了这么一句代码：

fileWriter = new FileWriter("/home/"+UUID.randomUUID());

然后在后面的代码中，往这个文件追加写入了运行结果。
此时我们需要到第二台或者第三台主机上查看这些文件。
切换到/home目录下

前三个文件就是以随机的uuid明明的记录文件，使用more+文件名命令打开看看。
More 命令一下。

确实有结果，程序跑成功了。

如果这个时候，我们不管程序，它就会继续跑，文件无限增大。所以在项目结束后，需要关闭这个拓扑的工作。

回到主节点上，切换到storm目录下的bin，运行以下命令：
./storm kill example_top
其中的example_top是拓扑名称，是我们在代码里面定义的，在UI上也可以看到。Kill完之后，去UI刷新一下，发现拓扑的状态变为KILLED，再过一会你再刷新，拓扑就直接消失了。

至此，一个完整的程序就跑完了。关于storm，就先告一段落了。

终于把配置部分写完了。其实如果你觉得浪费时间，不想熟悉linux，那么其实你可以根本就不看，直接用本地模式跑一个简单的程序也行。但是事实上，我们学习storm，就是为了让其处理实时的大数据，配置集群有利于我们更熟悉整个框架。好了不多说，进入正题，这篇文章，我将用简单的实例代码来初步接触storm。

在storm官网下载的压缩包里有官方给出的例子，建议大家根据这个来学习一下。由于里面涉及到maven，需要配置一些东西，所以这里暂时不直接跑它的例子。取而代之的，我们用一个比较直观的例子来熟悉storm。

需求如下：

从一个数据源中获取随机的单词，然后将其转换为大写，再加个后缀，写入本地文件。

在storm的框架下，我们可以这样来解决这个问题。首先因为我们没有数据源，所以我们用RandomSpout来模拟产生源源不断的数据，然后用UpperBolt来接收数据，将字母变为大写，再发送给SuffixBolt，给它加后缀，然后写入文件。

接下来就用代码来说话。

eclipse新建工程，并在build path中把storm的jar包引进来（通常在lib文件夹下）。

写第一个类RandomSpout，用于产生数据

import java.util.Map;
import java.util.Random;
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{
	SpoutOutputCollector collector=null;
	String[] words={"storm","apache","spark","hadoop"};
	@Override
	public void nextTuple() {
		Random random = new Random();
		String word =words[random.nextInt(words.length)];
		//发送消息
		collector.emit(new Values(word));
		
	}

	@Override
	public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector collector) {
		//初始化collector
		this.collector=collector;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("src_word"));
	}

}

要点：

• open()方法是运行的最早的一个方法，我们这里初始化了collector，以方便给别的方法使用。
• 最开始创建了一个String数组，作为单词源。里面有四个单词：storm,apache,spark,hadoop。
• nextTuple()方法会被循环调用，里面的collector.emit(new Values(word))用于发送数据给接收者。
• declareOutputFields()方法用来定义输出的字段。

然后新建UpperBolt类，用于接收RandomSpout发来的单词，并将单词转化为大写。

import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class UpperBolt extends BaseBasicBolt {

	@Override
	public void execute(Tuple tuple, BasicOutputCollector collector) {
		String src_word = tuple.getString(0);
		String upper_word=src_word.toUpperCase();
		//发送消息出去
		collector.emit(new Values(upper_word));
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("upper_word"));
	}

}

要点：

• execute()方法不断接收tuple，并在里面写逻辑进行处理，并通过collector将封装成tuple的数据emit出去
• tuple是storm传输数据的单位，我们需要将数据处理好之后封装成tuple，再发送出去，new Values(upper_word)就是封装成tuple的一种形式

再新建SuffixBolt类，用于接受经过大写处理的单词，同时给单词加上后缀“-handled”。最后再写入本地文件。

import java.io.FileWriter;
import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

public class SuffixBolt extends BaseBasicBolt{
	FileWriter fileWriter = null; 
	
    @Override
	public void prepare(Map stormConf, TopologyContext context) {
		try {
			fileWriter = new FileWriter("/home/"+UUID.randomUUID());//初始化fileWriter，写到本地文件，在/home/目录下，文件用随机的UUID命名
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	@Override
	public void execute(Tuple tuple, BasicOutputCollector collector) {
		String upper_word = tuple.getString(0);//接收数据
		String suffix_word = upper_word+"-handled";//加后缀
		
		try {
			fileWriter.append(suffix_word+"\n");//写入文件
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("suffix_word") );
	}

}

要点：

• execute()方法不断接收tuple，但和之前的Bolt不一样的是，这次我们不用发送数据了，而是将处理后的数据直接写入本地文件中。

最后，写个主程序类TopMain，用来串联所有的组件。

import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.topology.TopologyBuilder;

public class TopMain {

	public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
		
		//新建对象
		TopologyBuilder topologyBuilder = new TopologyBuilder();
		//设置spout，并行度为4，第一个参数为自定义的名字
		topologyBuilder.setSpout("randomspout", new RandomSpout(),4);
		//设置bolt，数据来源是上面的spout
		topologyBuilder.setBolt("upperbolt", new UpperBolt(),4).shuffleGrouping("randomspout");
		topologyBuilder.setBolt("suffixbolt", new SuffixBolt(), 4).shuffleGrouping("upperbolt");
		
		//创建拓扑
		StormTopology top = topologyBuilder.createTopology();
		
		//创建config，配置storm
		Config config = new Config();
		config.setNumWorkers(3);  //进程数
		config.setNumAckers(2);   //应答器数目

		//提交拓扑给集群
		StormSubmitter.submitTopology("example_top", config, top);
	
	}
}

要点：

• 拓扑是storm里面一个很重要的概念，说白了，就是工程的整合，在这里，你可以设置worker数，Acker数，串联Spout和Bolt等等，关于它的配置，其实有很多内容，这里先不多说，注释把基本的功能写出来了，其实每个地方都可以深入探讨的。

好，代码写完了，怎么跑呢，且听下回分解。

上一篇讲了jdk的安装，这一篇继续配置。在安装storm和zookeeper之前，我们需要安装一些依赖，否则在正式安装的时候会出错。安装依赖也非常简单，yum可以帮我们解决所有的问题。下面说说要安装什么依赖。

先安装c++编译器。

yum install gcc-c++

按照提示输入y，回车，安装完毕。

再安装软件包uuid-devel和libuuid-devel

yum install uuid-devel

yum install libuuid-devel

随后安装libtool

yum install libtool

安装完以上的所有，我们继续安装storm所需要的Zeromq和jzmq
zeromq下载地址：
http://download.zeromq.org/zeromq-4.1.3.tar.gz
jzmq下载地址：
https://codeload.github.com/zeromq/jzmq/zip/master

把下载的两个压缩包上传到服务器，并在服务器上解压。解压完成之后，切换到对应目录下，编译安装。

cd zeromq-4.1.3
./configure
make
make install

安装jzmq的过程也是如法炮制

cd jzmq-master
./autogen.sh
./configure
make
make install

至此，所需依赖就安装完了，下一篇，终于可以正式安装storm了。