Storm
1. 信息流处理{Stream processing}
Storm可用来实时处理新数据和更新数据库,兼具容错性和可扩展性。
2. 连续计算{Continuous computation}
Storm可进行连续查询并把结果即时反馈给客户端。比如把Twitter上的热门话题发送到浏览器中。
3. 分布式远程程序调用{Distributed RPC}
Storm可用来并行处理密集查询。Storm的拓扑结构是一个等待调用信息的分布函数,当它收到一条调用信息后,会对查询进行计算,并返回查询结果。 举个例子Distributed RPC可以做并行搜索或者处理大集合的数据。
Storm集群的组件介绍:
storm集群是表面上类似于Hadoop集群。而在Hadoop上运行“MapReduce作业”,在storm运行“topologies”。 “jobs”和“topologies”本身有很大的不同 - 一个关键的区别是,一个MapReduce作业最终完成,而一个永远的拓扑信息进行处理(或直到你杀了它)。
有两种类型的节点运行在Storm集群中:
(1)主节点(master)主节点运行着一个守护进程“Nimbus“,这个类似于hadoop的jobtracker,Nimbus 负责将任务分配给worker机器,并监视故障。
(2)工作节点(worker)运行着一个守护进程"Supervisor",他主管侦听分配给它的机器和启动工作,并停止工作进程在必要时依据"Nimbus"已经分配给它的任务。
Nimbus与supervisor之间的所有通信都是通过ZK来传递的,另外Nimbus的守护进程和监事守护进程快速失败和无状态的。所有的状态都是保存在ZK中或者本地的磁盘中的。这就意味着你可以通过Kill -9来杀死 Nimbus与supervisor进程,然后他们启动就像没发生过任何事情一样。这样的设计使得storm集群非常稳定。
如何将你的storm程序在storm集群中跑呢?
storm jar all-my-code.jar backtype.storm.MyTopology arg1 arg2
all-my-code.jar(包名) backtype.storm.MyTopology(类名-默认是定位到main方法) arg1 arg2(参数)
Streams
一个stream是一个没有边界的tuples.storm将一个流分成以分布式和可靠的方式的新的流。提供了一个数据流的转换:“spouts”,“bolts”。spouts和bolts有你实现运行的应用程序特定的逻辑接口。spouts其实就是一个数据源,一个spout去读取数据源,以tuples的形式通过emit()发射出去,然后bolts接受tuples。实际应用中可以有多个spout发射数据,并且由多个bolts接收数据,bolts也可以将接收到的数据再发射给其他bolts。
结构如下图所示:
这个拓扑将永远运行下去,或者直到你杀死它。storm将自动重新分配任何失败的任务。此外,Storm保证不会有数据丢失,即使宕机。
一个简单的例子拓扑:
1、SimpleSpout类继承BaseRichSpout类,用来产生数据并且向topology里面发出消息:tuple。
package com.ljq.helloword; 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; /** * Spout起到和外界沟通的作用,他可以从一个数据库中按照某种规则取数据,也可以从分布式队列中取任务 * * @author Administrator * */ @SuppressWarnings("serial") public class SimpleSpout extends BaseRichSpout{ //用来发射数据的工具类 private SpoutOutputCollector collector; private static String[] info = new String[]{ "comaple\t,12424,44w46,654,12424,44w46,654,", "lisi\t,435435,6537,12424,44w46,654,", "lipeng\t,45735,6757,12424,44w46,654,", "hujintao\t,45735,6757,12424,44w46,654,", "jiangmin\t,23545,6457,2455,7576,qr44453", "beijing\t,435435,6537,12424,44w46,654,", "xiaoming\t,46654,8579,w3675,85877,077998,", "xiaozhang\t,9789,788,97978,656,345235,09889,", "ceo\t,46654,8579,w3675,85877,077998,", "cto\t,46654,8579,w3675,85877,077998,", "zhansan\t,46654,8579,w3675,85877,077998,"}; Random random=new Random(); /** * 初始化collector */ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { this.collector = collector; } /** * 在SpoutTracker类中被调用,每调用一次就可以向storm集群中发射一条数据(一个tuple元组),该方法会被不停的调用 */ @Override public void nextTuple() { try { String msg = info[random.nextInt(11)]; // 调用发射方法 collector.emit(new Values(msg)); // 模拟等待100ms Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } } /** * 定义字段id,该id在简单模式下没有用处,但在按照字段分组的模式下有很大的用处。 * 该declarer变量有很大作用,我们还可以调用declarer.declareStream();来定义stramId,该id可以用来定义更加复杂的流拓扑结构 */ @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("source")); //collector.emit(new Values(msg));参数要对应 } }
2、SimpleBolt类继承BaseBasicBolt类,处理一个输入tuple。
package com.ljq.helloword; 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; /** * 接收喷发节点(Spout)发送的数据进行简单的处理后,发射出去。 * * @author Administrator * */ @SuppressWarnings("serial") public class SimpleBolt extends BaseBasicBolt { public void execute(Tuple input, BasicOutputCollector collector) { try { String msg = input.getString(0); if (msg != null){ //System.out.println("msg="+msg); collector.emit(new Values(msg + "msg is processed!")); } } catch (Exception e) { e.printStackTrace(); } } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("info")); } }
3、SimpleTopology类包含一个main函数,是Storm程序执行的入口点,包括一个数据喷发节点spout和一个数据处理节点bolt。
package com.ljq.helloword; import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.StormSubmitter; import backtype.storm.topology.TopologyBuilder; /** * 定义了一个简单的topology,包括一个数据喷发节点spout和一个数据处理节点bolt。 * * @author Administrator * */ public class SimpleTopology { public static void main(String[] args) { try { // 实例化TopologyBuilder类。 TopologyBuilder topologyBuilder = new TopologyBuilder(); // 设置喷发节点并分配并发数,该并发数将会控制该对象在集群中的线程数。 topologyBuilder.setSpout("SimpleSpout", new SimpleSpout(), 1); // 设置数据处理节点并分配并发数。指定该节点接收喷发节点的策略为随机方式。 topologyBuilder.setBolt("SimpleBolt", new SimpleBolt(), 3).shuffleGrouping("SimpleSpout"); Config config = new Config(); config.setDebug(true); if (args != null && args.length > 0) { config.setNumWorkers(1); StormSubmitter.submitTopology(args[0], config, topologyBuilder.createTopology()); } else { // 这里是本地模式下运行的启动代码。 config.setMaxTaskParallelism(1); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("simple", config, topologyBuilder.createTopology()); } } catch (Exception e) { e.printStackTrace(); } } }
运行结果效果如下:
相关推荐
Storm是目前流行的分布式实时流计算框架之一,它提供了可容错分布式计算所要求的基本需求和保障机制,可以满足高吞吐 ,实时的关键业务应用的需求。在编写基于Storm的分布式计算应用时,我们首先需要创建一个拓扑...
它主要参考当前流行的实时计算框架Storm的任务分发和Spark Streaming的Mini-Batch处理思想设计,设计目的是为了降低当前大数据时代的分布式实时计算入门门槛,方便初中级学者上手,快速实现企业所需计算实时性要求...
开源分布式实时计算框架Storm在互联网、金融、电子商务等领域得到了广泛应用。Storm默认采用轮询的调度策略,且依赖用户对topology任务的并行度配置,当配置不合理时依然会造成topology处理时延增大、吞吐量降低等...
storm是一个类似于Hadoop勺实时数据处理框架,也是一个非常有效的开源实时计算工具,通常被比作“实时的Hadoop”。 《大数据技术丛书:Storm实时数据处理》通过丰富的实例,系统讲解Storm的基础知识和实时数据处理...
1.1简介Storm 是一个开源的分布式实时计算框架,可以以简单、可靠的方式进行大数据流的处理。通常用于实时分析,在线机器学习、持续计算、分布式 RPC、ETL
亿矿云大数据处理框架:借助Hadoop、Spark、Storm等分布式处理架构,满足海量数据的批处理和流处理计算需求
【资源说明】 1、该资源包括项目的全部源码,下载可以直接使用! 2、本项目适合作为计算机、数学、电子信息等专业的课程设计、期末大...基于Hadoop、Spark、Storm等分布式处理架构的大数据处理框架源码+项目说明.zip
Storm是一个分布式的实时计算框架,可以很方便地对流式数据进行实时处理和分析,能运用在实时分析、在线数据挖掘、持续计算以及分布式RPC等场景下。Storm的实时性可以使得数据从收集到处理展示在秒级别内完成,从而...
Redis 开源的⽀持⽹络,基于内存可持久化⽇志,key-value数据库,可⽤于 数据库 缓存 消息中间件 Neo4j 开源⾼性能的NoSQL图形数据库 7 数据处理 MapReduce 分布式离线的计算框架 批处理 ⽇渐被spark和flink取代 ...
Samza是一款优秀的分布式流处理框架,非常像Twitter的流处理系统Storm(即Lambda架构Hadoop+Storm)。不同的是Samza基于Hadoop,而且使用了LinkedIn自家的Kafka分布式消息系统(即Kappa架构Kafka+Samza)。其优势...
Storm是一个免费开源、分布式、高容错的实时计算系统。Storm令持续不断的流计算变得容易,弥补了Hadoop批处理所不能满足的实时要求。Storm是一个服务框架,支持热部署,即时上线或下线App。 你可以在Storm之上使用各种...
针对社交媒体数据的特点及其分析的挑战性,提出了一种基于实时计算框架Storm、批处理框架Hadoop和高效可水平扩展的NoSQL数据库MongoDB的分布式社交媒体数据处理方案,并依此指导实现基于Twitter流式数据的流感疫情...
这种系统结合了Hadoop分布式计算框架的能力,可处理庞大的游戏数据集,并提供多维度的数据分析功能。以下是这种系统的主要特点和功能: 1. 数据采集与存储:系统通过Hadoop的分布式存储组件HDFS(Hadoop ...
越来越多的分布式计算开始强依赖ZK,比如Storm、Hbase Zookeeper对分布式开发带来很多便利,用ZK的独有特性巧妙地解决了很多难题; 很多分布式技术用到Zookeeper或多或少特性,尤其是新生代分布式技术几乎都会依赖...
数 据 层 结构化业务数据、机器数据 半结构化数据、机器数据 序列化 算法库 机器学习 Storm内存 流式计算框架 Hadoop MapReduce 计算框架 Spark 并行计算框架 计 算 层 运营 分析 日志 分析 个性化 推荐 供应链 分析...
被誉为最火的流式处理框架,弥补了Hadoop的众多缺点,Storm经常用于在实时分析、在线机器学习、持续计算、分布式远程调用和ETL等领域。本文介绍的是基于Storm的Nginxlog实时监控系统。Hadoop的缺点也和它的优点同样...
此外还有针对实时在线流式数据处理方面的,同样也是分布式的计算框架Storm,也能很好的满足数据实时性分析、处理的要求。最后还有SpringBatch,这个完全面向批处理的框架,可以大规模的应用于企业级的
BigdataNote -----------欢迎访问------------ 我的个人博客: 我的个人公众号:GoAl分享 大数据优秀博客推荐: 过往记忆大数据博客: 董西城博客: lwx的数据田地: 美团技术团队: 林子雨厦门大学实验室: ...