Storm 入门的Demo教程
2018-03-16 20:03
351 查看
Storm介绍
Storm是Twitter开源的分布式实时大数据处理框架,最早开源于github,从0.9.1版本之后,归于Apache社区,被业界称为实时版Hadoop。随着越来越多的场景对Hadoop的MapReduce高延迟无法容忍,比如网站统计、推荐系统、预警系统、金融系统(高频交易、股票)等等,大数据实时处理解决方案(流计算)的应用日趋广泛,目前已是分布式技术领域最新爆发点,而Storm更是流计算技术中的佼佼者和主流。Storm的核心组件
Nimbus:即Storm的Master,负责资源分配和任务调度。一个Storm集群只有一个Nimbus。Supervisor:即Storm的Slave,负责接收Nimbus分配的任务,管理所有Worker,一个Supervisor节点中包含多个Worker进程。
Worker:工作进程,每个工作进程中都有多个Task。
Task:任务,在 Storm 集群中每个 Spout 和 Bolt 都由若干个任务(tasks)来执行。每个任务都与一个执行线程相对应。
Topology:计算拓扑,Storm 的拓扑是对实时计算应用逻辑的封装,它的作用与 MapReduce 的任务(Job)很相似,区别在于 MapReduce 的一个 Job 在得到结果之后总会结束,而拓扑会一直在集群中运行,直到你手动去终止它。拓扑还可以理解成由一系列通过数据流(Stream Grouping)相互关联的 Spout 和 Bolt 组成的的拓扑结构。
Stream:数据流(Streams)是 Storm 中最核心的抽象概念。一个数据流指的是在分布式环境中并行创建、处理的一组元组(tuple)的无界序列。数据流可以由一种能够表述数据流中元组的域(fields)的模式来定义。
Spout:数据源(Spout)是拓扑中数据流的来源。一般 Spout 会从一个外部的数据源读取元组然后将他们发送到拓扑中。根据需求的不同,Spout 既可以定义为可靠的数据源,也可以定义为不可靠的数据源。一个可靠的 Spout能够在它发送的元组处理失败时重新发送该元组,以确保所有的元组都能得到正确的处理;相对应的,不可靠的 Spout 就不会在元组发送之后对元组进行任何其他的处理。一个 Spout可以发送多个数据流。
Bolt:拓扑中所有的数据处理均是由 Bolt 完成的。通过数据过滤(filtering)、函数处理(functions)、聚合(aggregations)、联结(joins)、数据库交互等功能,Bolt 几乎能够完成任何一种数据处理需求。一个 Bolt 可以实现简单的数据流转换,而更复杂的数据流变换通常需要使用多个 Bolt 并通过多个步骤完成。
Stream grouping:为拓扑中的每个 Bolt 的确定输入数据流是定义一个拓扑的重要环节。数据流分组定义了在 Bolt 的不同任务(tasks)中划分数据流的方式。在 Storm 中有八种内置的数据流分组方式。
Reliability:可靠性。Storm 可以通过拓扑来确保每个发送的元组都能得到正确处理。通过跟踪由 Spout 发出的每个元组构成的元组树可以确定元组是否已经完成处理。每个拓扑都有一个“消息延时”参数,如果 Storm 在延时时间内没有检测到元组是否处理完成,就会将该元组标记为处理失败,并会在稍后重新发送该元组。
Storm程序再Storm集群中运行的示例图如下:
Topology
为什么把Topology单独提出来呢,因为Topology是我们开发程序主要的用的组件。Topology和MapReduce很相像。
MapReduce是Map进行获取数据,Reduce进行处理数据。
而Topology则是使用Spout获取数据,Bolt来进行计算。
总的来说就是一个Topology由一个或者多个的Spout和Bolt组成。
具体流程是怎么走,可以通过查看下面这张图来进行了解。
示例图:
注:图片来源http://www.tianshouzhi.com/api/tutorials/storm/52。
图片有三种模式,解释如下:
第一种比较简单,就是由一个Spout获取数据,然后交给一个Bolt进行处理;
第二种稍微复杂点,由一个Spout获取数据,然后交给一个Bolt进行处理一部分,然后在交给下一个Bolt进行处理其他部分。
第三种则比较复杂,一个Spout可以同时发送数据到多个Bolt,而一个Bolt也可以接受多个Spout或多个Bolt,最终形成多个数据流。但是这种数据流必须是有方向的,有起点和终点,不然会造成死循环,数据永远也处理不完。就是Spout发给Bolt1,Bolt1发给Bolt2,Bolt2又发给了Bolt1,最终形成了一个环状。
Storm 集群安装
之前已经写过了,这里就不在说明了。博客地址:http://www.panchengming.com/2018/01/26/pancm70/
Storm Hello World
前面讲了一些Storm概念,可能在理解上不太清楚,那么这里我们就用一个Hello World代码示例来体验下Storm运作的流程吧。环境准备
在进行代码开发之前,首先得做好相关的准备。本项目是使用Maven构建的,使用Storm的版本为1.1.1。
Maven的相关依赖如下:
<!--storm相关jar --> <dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-core</artifactId> <version>1.1.1</version> <scope>provided</scope> </dependency>
具体流程
在写代码的时候,我们先来明确要用Storm做什么。那么第一个程序,就简单的输出下信息。
具体步骤如下:
启动topology,设置好Spout和Bolt。
将Spout获取的数据传递给Bolt。
Bolt接受Spout的数据进行打印。
Spout
那么首先开始编写Spout类。一般是实现 IRichSpout 或继承BaseRichSpout该类,然后实现该方法。这里我们继承BaseRichSpout这个类,该类需要实现这几个主要的方法:
一、open
open()方法中是在ISpout接口中定义,在Spout组件初始化时被调用。
有三个参数,它们的作用分别是:
Storm配置的Map;
topology中组件的信息;
发射tuple的方法;
代码示例:
@Override public void open(Map map, TopologyContext arg1, SpoutOutputCollector collector) { System.out.println("open:"+map.get("test")); this.collector = collector; }
二、nextTuple
nextTuple()方法是Spout实现的核心。
也就是主要执行方法,用于输出信息,通过
collector.emit方法发射。
这里我们的数据信息已经写死了,所以这里我们就直接将数据进行发送。
这里设置只发送两次。
代码示例:
@Override public void nextTuple() { if(count<=2){ System.out.println("第"+count+"次开始发送数据..."); this.collector.emit(new Values(message)); } count++; }
三、declareOutputFields
declareOutputFields是在IComponent接口中定义,用于声明数据格式。
即输出的一个Tuple中,包含几个字段。
因为这里我们只发射一个,所以就指定一个。如果是多个,则用逗号隔开。
代码示例:
@Override public void declareOutputFields(OutputFieldsDeclarer declarer) { System.out.println("定义格式..."); declarer.declare(new Fields(field)); }
四、ack
ack是在ISpout接口中定义,用于表示Tuple处理成功。
代码示例:
@Override public void ack(Object obj) { System.out.println("ack:"+obj); }
五、fail
fail是在ISpout接口中定义,用于表示Tuple处理失败。
代码示例:
@Override public void fail(Object obj) { System.out.println("失败:"+obj); }
六、close
close是在ISpout接口中定义,用于表示Topology停止。
代码示例:
@Override public void close() { System.out.println("关闭..."); }
至于还有其他的,这里就不在一一列举了。
Bolt
Bolt是用于处理数据的组件,主要是由execute方法来进行实现。一般来说需要实现 IRichBolt 或继承BaseRichBolt该类,然后实现其方法。需要实现方法如下:
一、prepare
在Bolt启动前执行,提供Bolt启动环境配置的入口。
参数基本和Sqout一样。
一般对于不可序列化的对象进行实例化。
这里的我们就简单的打印下
@Override public void prepare(Map map, TopologyContext arg1, OutputCollector collector) { System.out.println("prepare:"+map.get("test")); this.collector=collector; }
注:如果是可以序列化的对象,那么最好是使用构造函数。
二、execute
execute()方法是Bolt实现的核心。
也就是执行方法,每次Bolt从流接收一个订阅的tuple,都会调用这个方法。
从tuple中获取消息可以使用
tuple.getString()和
tuple.getStringByField();这两个方法。个人推荐第二种,可以通过field来指定接收的消息。
注:如果继承的是IRichBolt,则需要手动ack。这里就不用了,BaseRichBolt会自动帮我们应答。
代码示例:
@Override public void execute(Tuple tuple) { // String msg=tuple.getString(0); String msg=tuple.getStringByField("test"); //这里我们就不做消息的处理,只打印 System.out.println("Bolt第"+count+"接受的消息:"+msg); count++; /** * * 没次调用处理一个输入的tuple,所有的tuple都必须在一定时间内应答。 * 可以是ack或者fail。否则,spout就会重发tuple。 */ // collector.ack(tuple); }
三、declareOutputFields
和Spout的一样。
因为到了这里就不再输出了,所以就什么都没写。
@Override public void declareOutputFields(OutputFieldsDeclarer arg0) { }
cleanup
cleanup是IBolt接口中定义,用于释放bolt占用的资源。
Storm在终止一个bolt之前会调用这个方法。
因为这里没有什么资源需要释放,所以就简单的打印一句就行了。
@Override public void cleanup() { System.out.println("资源释放"); }
Topology
这里我们就是用main方法进行提交topology。不过在提交topology之前,需要进行相应的设置。
这里我就不一一细说了,代码的注释已经很详细了。
代码示例:
import org.apache.storm.Config; import org.apache.storm.LocalCluster; import org.apache.storm.StormSubmitter; import org.apache.storm.topology.TopologyBuilder; /** * * Title: App * Description: * storm测试 * Version:1.0.0 * @author pancm * @date 2018年3月6日 */ public class App { private static final String str1="test1"; private static final String str2="test2"; public static void main(String[] args) { // TODO Auto-generated method stub //定义一个拓扑 TopologyBuilder builder=new TopologyBuilder(); //设置一个Executeor(线程),默认一个 builder.setSpout(str1, new TestSpout()); //设置一个Executeor(线程),和一个task builder.setBolt(str2, new TestBolt(),1).setNumTasks(1).shuffleGrouping(str1); Config conf = new Config(); conf.put("test", "test"); try{ //运行拓扑 if(args !=null&&args.length>0){ //有参数时,表示向集群提交作业,并把第一个参数当做topology名称 System.out.println("远程模式"); StormSubmitter.submitTopology(args[0], conf, builder.createTopology()); } else{//没有参数时,本地提交 //启动本地模式 System.out.println("本地模式"); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("111" ,conf, builder.createTopology() ); Thread.sleep(10000); // 关闭本地集群 cluster.shutdown(); } }catch (Exception e){ e.printStackTrace(); } } }
运行该方法,输出结果如下:
本地模式 定义格式... open:test 第1次开始发送数据... 第2次开始发送数据... prepare:test Bolt第1接受的消息:这是个测试消息! Bolt第2接受的消息:这是个测试消息! 资源释放 关闭...
到这里,是不是基本上对Storm的运作有些了解了呢。
这个demo达到了上述的三种模式图中的第一种,一个Spout传输数据, 一个Bolt处理数据。
那么如果我们想达到第二种模式呢,那又该如何做呢?
假如我们想统计下在一段文本中的单词出现频率的话,我们只需执行一下步骤就可以了。
1.首先将Spout中的message消息进行更改为数组,并依次将消息发送到TestBolt。
2.然后TestBolt将获取的数据进行分割,将分割的数据发送到TestBolt2。
3.TestBolt2对数据进行统计,在程序关闭的时候进行打印。
4.Topology成功配置并且启动之后,等待20秒左右,关闭程序,然后得到输出的结果。
代码示例如下:
Spout
用于发送消息。
import java.util.Map; import org.apache.storm.spout.SpoutOutputCollector; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; import org.apache.storm.topology.base.BaseRichSpout; import org.apache.storm.tuple.Fields; import org.apache.storm.tuple.Values; /** * * Title: TestSpout * Description: * 发送信息 * Version:1.0.0 * @author pancm * @date 2018年3月6日 */ public class TestSpout extends BaseRichSpout{ private static final long serialVersionUID = 225243592780939490L; private SpoutOutputCollector collector; private static final String field="word"; private int count=1; private String[] message = { "My nickname is xuwujing", "My blog address is http://www.panchengming.com/", "My interest is playing games" }; /** * open()方法中是在ISpout接口中定义,在Spout组件初始化时被调用。 * 有三个参数: * 1.Storm配置的Map; * 2.topology中组件的信息; * 3.发射tuple的方法; */ @Override public void open(Map map, TopologyContext arg1, SpoutOutputCollector collector) { System.out.println("open:"+map.get("test")); this.collector = collector; } /** * nextTuple()方法是Spout实现的核心。 * 也就是主要执行方法,用于输出信息,通过collector.emit方法发射。 */ @Override public void nextTuple() { if(count<=message.length){ System.out.println("第"+count +"次开始发送数据..."); this.collector.emit(new Values(message[count-1])); } count++; } /** * declareOutputFields是在IComponent接口中定义,用于声明数据格式。 * 即输出的一个Tuple中,包含几个字段。 */ @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { System.out.println("定义格式..."); declarer.declare(new Fields(field)); } /** * 当一个Tuple处理成功时,会调用这个方法 */ @Override public void ack(Object obj) { System.out.println("ack:"+obj); } /** * 当Topology停止时,会调用这个方法 */ @Override public void close() { System.out.println("关闭..."); } /** * 当一个Tuple处理失败时,会调用这个方法 */ @Override public void fail(Object obj) { System.out.println("失败:"+obj); } }
TestBolt
用于分割单词。
import java.util.Map; import org.apache.storm.task.OutputCollector; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; import org.apache.storm.topology.base.BaseRichBolt; import org.apache.storm.tuple.Fields; import org.apache.storm.tuple.Tuple; import org.apache.storm.tuple.Values; /** * * Title: TestBolt * Description: * 对单词进行分割 * Version:1.0.0 * @author pancm * @date 2018年3月16日 */ public class TestBolt extends BaseRichBolt{ /** * */ private static final long serialVersionUID = 4743224635827696343L; private OutputCollector collector; /** * 在Bolt启动前执行,提供Bolt启动环境配置的入口 * 一般对于不可序列化的对象进行实例化。 * 注:如果是可以序列化的对象,那么最好是使用构造函数。 */ @Override public void prepare(Map map, TopologyContext arg1, OutputCollector collector) { System.out.println("prepare:"+map.get("test")); this.collector=collector; } /** * execute()方法是Bolt实现的核心。 * 也就是执行方法,每次Bolt从流接收一个订阅的tuple,都会调用这个方法。 */ @Override public void execute(Tuple tuple) { String msg=tuple.getStringByField("word"); System.out.println("开始分割单词:"+msg); String[] words = msg.toLowerCase().split(" "); for (String word : words) { this.collector.emit(new Values(word));//向下一个bolt发射数据 } } /** * 声明数据格式 */ @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("count")); } /** * cleanup是IBolt接口中定义,用于释放bolt占用的资源。 * Storm在终止一个bolt之前会调用这个方法。 */ @Override public void cleanup() { System.out.println("TestBolt的资源释放"); } }
Test2Bolt
用于统计单词出现次数。
import java.util.HashMap; import java.util.Map; import org.apache.storm.task.OutputCollector; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; import org.apache.storm.topology.base.BaseRichBolt; import org.apache.storm.tuple.Tuple; /** * * Title: Test2Bolt * Description: * 统计单词出现的次数 * Version:1.0.0 * @author pancm * @date 2018年3月16日 */ public class Test2Bolt extends BaseRichBolt{ /** * */ private static final long serialVersionUID = 4743224635827696343L; /** * 保存单词和对应的计数 */ private HashMap<String, Integer> counts = null; private long count=1; /** * 在Bolt启动前执行,提供Bolt启动环境配置的入口 * 一般对于不可序列化的对象进行实例化。 * 注:如果是可以序列化的对象,那么最好是使用构造函数。 */ @Override public void prepare(Map map, TopologyContext arg1, OutputCollector collector) { System.out.println("prepare:"+map.get("test")); this.counts=new HashMap<String, Integer>(); } /** * execute()方法是Bolt实现的核心。 * 也就是执行方法,每次Bolt从流接收一个订阅的tuple,都会调用这个方法。 * */ @Override public void execute(Tuple tuple) { String msg=tuple.getStringByField("count"); System.out.println("第"+count+"次统计单词出现的次数"); /** * 如果不包含该单词,说明在该map是第一次出现 * 否则进行加1 */ if (!counts.containsKey(msg)) { counts.put(msg, 1); } else { counts.put(msg, counts.get(msg)+1); } count++; } /** * cleanup是IBolt接口中定义,用于释放bolt占用的资源。 * Storm在终止一个bolt之前会调用这个方法。 */ @Override public void cleanup() { System.out.println("===========开始显示单词数量============"); for (Map.Entry<String, Integer> entry : counts.entrySet()) { System.out.println(entry.getKey() + ": " + entry.getValue()); } System.out.println("===========结束============"); System.out.println("Test2Bolt的资源释放"); } /** * 声明数据格式 */ @Override public void declareOutputFields(OutputFieldsDeclarer arg0) { } }
Topology
主程序入口。
import org.apache.storm.Config; import org.apache.storm.LocalCluster; import org.apache.storm.StormSubmitter; import org.apache.storm.topology.TopologyBuilder; import org.apache.storm.tuple.Fields; /** * * Title: App * Description: * storm测试 * Version:1.0.0 * @author pancm * @date 2018年3月6日 */ public class App { private static final String test_spout="test_spout"; private static final String test_bolt="test_bolt"; private static final String test2_bolt="test2_bolt"; public static void main(String[] args) { //定义一个拓扑 TopologyBuilder builder=new TopologyBuilder(); //设置一个Executeor(线程),默认一个 builder.setSpout(test_spout, new TestSpout(),1); //shuffleGrouping:表示是随机分组 //设置一个Executeor(线程),和一个task builder.setBolt(test_bolt, new TestBolt(),1).setNumTasks(1).shuffleGrouping(test_spout); //fieldsGrouping:表示是按字段分组 //设置一个Executeor(线程),和一个task builder.setBolt(test2_bolt, new Test2Bolt(),1).setNumTasks(1).fieldsGrouping(test_bolt, new Fields("count")); Config conf = new Config(); conf.put("test", "test"); try{ //运行拓扑 if(args !=null&&args.length>0){ //有参数时,表示向集群提交作业,并把第一个参数当做topology名称 System.out.println("运行远程模式"); StormSubmitter.submitTopology(args[0], conf, builder.createTopology()); } else{//没有参数时,本地提交 //启动本地模式 System.out.println("运行本地模式"); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("Word-counts" ,conf, builder.createTopology() ); Thread.sleep(20000); // //关闭本地集群 cluster.shutdown(); } }catch (Exception e){ e.printStackTrace(); } } }
输出结果:
运行本地模式 定义格式... open:test 第1次开始发送数据... 第2次开始发送数据... 第3次开始发送数据... prepare:test prepare:test 开始分割单词:My nickname is xuwujing 开始分割单词:My blog address is http://www.panchengming.com/ 开始分割单词:My interest is playing games 第1次统计单词出现的次数 第2次统计单词出现的次数 第3次统计单词出现的次数 第4次统计单词出现的次数 第5次统计单词出现的次数 第6次统计单词出现的次数 第7次统计单词出现的次数 第8次统计单词出现的次数 第9次统计单词出现的次数 第10次统计单词出现的次数 第11次统计单词出现的次数 第12次统计单词出现的次数 第13次统计单词出现的次数 第14次统计单词出现的次数 ===========开始显示单词数量============ address: 1 interest: 1 nickname: 1 games: 1 is: 3 xuwujing: 1 playing: 1 my: 3 blog: 1 http://www.panchengming.com/: 1 ===========结束============ Test2Bolt的资源释放 TestBolt的资源释放 关闭...
上述的是本地模式运行,如果想在Storm集群中进行使用,只需要将程序打包为jar,然后将程序上传到storm集群中,
输入:
storm jar xxx.jar xxx xxx
说明:第一个xxx是storm程序打包的包名,第二个xxx是运行主程序的路径,第三个xxx则表示主程序输入的参数,这个可以随意。
如果是使用maven打包的话,则需要在pom.xml加上
<plugin> <artifactId>maven-assembly-plugin</artifactId> <configuration> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> <archive> <manifest> <mainClass>com.pancm.storm.App</mainClass> </manifest> </archive> </configuration> </plugin>
成功运行程序之后,可以在Storm集群的UI界面查看该程序的状态。
到此,本文结束,谢谢阅读!
相关文章推荐
- Storm入门教程 第二章 构建Topology
- storm入门教程 第一章 前言[转]
- Storm入门教程:消息的可靠处理
- storm从入门到放弃教程(1)--storm初识
- Storm入门教程 第三章 Storm安装部署步骤
- Storm入门教程 第二章 构建Topology
- Storm学习入门视频教程
- RabbitMQ入门教程之二--》spring+rabbitmq简单demo
- 支付宝小程序入门教程+案例demo
- Storm入门教程:一致性事务
- Storm入门(十三)Storm Trident 教程
- storm入门教程 第一章 前言
- Storm入门教程 第二章 构建Topology
- Storm入门教程 第二章 构建Topology[转]
- 微信小程序入门教程+案例demo
- 无废话WCF入门教程六[一个简单的Demo]
- Storm入门教程:前言
- 最详细的Storm入门教程(二)
- storm入门教程 第一章 前言
- Storm入门教程 第三章Storm集群安装部署步骤[转]