storm研究与介绍课件

研究与介绍Storm目录2、Storm的组件1、Storm是什么？4、项目架构3、Storm应用场景1

.

Storm是什么？Master对Map的预处理：对于大量的数据进行切分，

划分为M个16~64M的数据分片(可通过参数自定义分片大小)调用Mapper函数：Master为Worker分配Map任务，

每个分片都对应一个Worker进行处理。各个Worker读取

并调用用户定义的Mapper函数处理数据，并将结果存入

HDFS，返回存储位置给Master。

一个Worker在Map阶段完成时，在HDFS中，

生成一个排好序的Key-values组成的文件。并将位置信息汇报给Master。

先来聊聊我们熟悉的Hadoop全量数据处理使用的大多是鼎鼎大名的hadoop或者hive，作为一个批处理系统，hadoop以其吞吐量大、自动容错等优点，在海量数据处理上得到了广泛的使用。Hadoop下的Map/Reduce框架对于数据的处理流程是：1、将要处理的数据上传到Hadoop的文件系统HDFS中。2、Map阶段以上的流程，粗略概括，就是从HDFS中获取数据，将其按照大小分片，进行分布式处理，最终输出结果。从流程来看，Hadoop框架进行数据处理有以下要求：1、数据已经存在在HDFS当中。2、数据间是少关联的。各个任务执行器在执行负责的数据时，无需考虑对其他数据

的影响，数据之间应尽可能是无联系、不会影响的。使用Hadoop，适合大批量

的数据处理，这是他所擅长的。由于基于Map/Reduce这种单级的

数据处理模型进行，因此，如果数据间的关联系较大，

需要进行数据的多级交互处理（某个阶段的处理数据依赖于上

一个阶段），需要进行多次map/reduce。又由于map/reduce

每次执行都需要遍历整个数据集，对于数据的实时计算

并不合适，于是有了storm3、Reduce阶段a)Master对Reduce的预处理：Master为Worker分配Reduce任务，他会将所有Mapper产生的数据进行映射，将相同key的任务分配给某个Worker。b)调用Reduce函数：各个Worker将分配到的数据集进行排序（使用工具类Merg），并调用用户自定义的Reduce函数，并将结果写入HDFS。每个Worker的Reduce任务完成后，都会在HDFS中生成一个输出文件。Hadoop并不将这些文件合并，因为这些文件往往会作为另一个Map/reduce程序的输入。那么问题来了，What

is

Storm?首先我们打开官网，可以看到：2.Storm的组件Storm集群和Hadoop集群表面上看很类似。但是Hadoop上运行的是MapReducejobs，而在Storm上运行的是拓扑（topology），这两者之间是非常不一样的。一个关键的区别是：一个MapReducejob最终会结束，而一个topology永远会运行（除非你手动kill掉）。Storm集群主要由一个主节点（Nimbus后台程序）和一群工作节点（workernode）Supervisor的节点组成，通过Zookeeper进行协调。Nimbus类似Hadoop里面的JobTracker。Nimbus负责在集群里面分发代码，分配计算任务给机器，并且监控状态。每一个工作节点上面运行一个叫做Supervisor的节点。Supervisor会监听分配给它那台机器的工作，根据需要启动/关闭工作进程。每一个工作进程执行一个topology的一个子集；一个运行的topology由运行在很多机器上的很多工作进程组成。Topology示意图：1、Nimbus主节点：主节点通常运行一个后台程序——Nimbus，用于响应分布在集群中的节点，分配任务和监测故障。这个很类似于Hadoop中的JobTracker。2、Supervisor工作节点：工作节点同样会运行一个后台程序——Supervisor，用于收听工作指派并基于要求运行工作进程。每个工作节点都是topology中一个子集的实现。而Nimbus和Supervisor之间的协调则通过Zookeeper系统或者集群。3、Zookeeper

Zookeeper是完成Supervisor和Nimbus之间协调的服务。而应用程序实现实时的逻辑则被封装进Storm中的“topology”。topology则是一组由Spouts（数据源）和Bolts（数据操作）通过StreamGroupings进行连接的图。下面对出现的术语进行更深刻的解析。4、Worker：运行具体处理组件逻辑的进程。5、Task：

worker中每一个spout/bolt的线程称为一个task.在storm0.8之后，task不再与物理线程对应，同一个spout/bolt的task可能会共享一个物理线程，该线程称为executor。6、Topology（拓扑）：

storm中运行的一个实时应用程序，因为各个组件间的消息流动形成逻辑上的一个拓扑结构。一个topology是spouts和bolts组成的图，通过streamgroupings将图中的spouts和bolts连接起来，如下图：一个topology会一直运行直到你手动kill掉，Storm自动重新分配执行失败的任务，并且Storm可以保证你不会有数据丢失（如果开启了高可靠性的话）。如果一些机器意外停机它上面的所有任务会被转移到其他机器上。运行一个topology很简单。首先，把你所有的代码以及所依赖的jar打进一个jar包。然后运行类似下面的这个命令：

stormjarall-my-code.jar

backtype.storm.MyTopologyarg1arg2这个命令会运行主类:backtype.strom.MyTopology,参数是arg1,arg2。这个类的main函数定义这个topology并且把它提交给Nimbus。stormjar负责连接到Nimbus并且上传jar包。Topology的定义是一个Thrift结构，并且Nimbus就是一个Thrift服务，你可以提交由任何语言创建的topology。上面的方面是用JVM-based语言提交的最简单的方法。7、Spout：消息源spout是Storm里面一个topology里面的消息生产者。简而言之，Spout从来源处读取数据并放入topology。Spout分成可靠和不可靠两种；当Storm接收失败时，可靠的Spout会对tuple（元组，数据项组成的列表）进行重发；而不可靠的Spout不会考虑接收成功与否只发射一次。消息源可以发射多条消息流stream。使用OutputFieldsDeclarer.declareStream来定义多个stream，然后使用SpoutOutputCollector来发射指定的stream。而Spout中最主要的方法就是nextTuple（），该方法会发射一个新的tuple到topology，如果没有新tuple发射则会简单的返回。要注意的是nextTuple方法不能阻塞，因为storm在同一个线程上面调用所有消息源spout的方法。另外两个比较重要的spout方法是ack和fail。storm在检测到一个tuple被整个topology成功处理的时候调用ack，否则调用fail。storm只对可靠的spout调用ack和fail。8、Bolt：

Topology中所有的处理都由Bolt完成。即所有的消息处理逻辑被封装在bolts里面。Bolt可以完成任何事，比如：连接的过滤、聚合、访问文件/数据库、等等。

Bolt从Spout中接收数据并进行处理，如果遇到复杂流的处理也可能将tuple发送给另一个Bolt进行处理。即需要经过很多blots。比如算出一堆图片里面被转发最多的图片就至少需要两步：第一步算出每个图片的转发数量。第二步找出转发最多的前10个图片。（如果要把这个过程做得更具有扩展性那么可能需要更多的步骤）。

Bolts可以发射多条消息流，使用OutputFieldsDeclarer.declareStream定义stream，使用OutputCollector.emit来选择要发射的stream。而Bolt中最重要的方法是execute（），以新的tuple作为参数接收。不管是Spout还是Bolt，如果将tuple发射成多个流，这些流都可以通过declareStream（）来声明。

bolts使用OutputCollector来发射tuple，bolts必须要为它处理的每一个tuple调用OutputCollector的ack方法，以通知Storm这个tuple被处理完成了，从而通知这个tuple的发射者spouts。一般的流程是：bolts处理一个输入tuple,发射0个或者多个tuple,然后调用ack通知storm自己已经处理过这个tuple了。storm提供了一个IBasicBolt会自动调用ack。9、Tuple：一次消息传递的基本单元。本来应该是一个key-value的map，但是由于各个组件间传递的tuple的字段名称已经事先定义好，所以tuple中只要按序填入各个value就行了，所以就是一个valuelist.10、Stream：源源不断传递的tuple就组成了stream。消息流stream是storm里的关键抽象。一个消息流是一个没有边界的tuple序列，而这些tuple序列会以一种分布式的方式并行地创建和处理。通过对stream中tuple序列中每个字段命名来定义stream。在默认的情况下，tuple的字段类型可以是：integer，long，short，byte，string，double，float，boolean和bytearray。你也可以自定义类型（只要实现相应的序列化器）。每个消息流在定义的时候会被分配给一个id，因为单向消息流使用的相当普遍，OutputFieldsDeclarer定义了一些方法让你可以定义一个stream而不用指定这个id。在这种情况下这个stream会分配个值为‘default’默认的id。

Storm提供的最基本的处理stream的原语是spout和bolt。你可以实现spout和bolt提供的接口来处理你的业务逻辑。11、StreamGroupings：StreamGrouping定义了一个流在Bolt任务间该如何被切分。这里有Storm提供的6个StreamGrouping类型：1）.随机分组（Shufflegrouping）：随机分发tuple到Bolt的任务，保证每个任务获得相等数量的tuple。2）.字段分组（Fieldsgrouping）：根据指定字段分割数据流，并分组。例如，根据“user-id”字段，相同“user-id”的元组总是分发到同一个任务，不同“user-id”的元组可能分发到不同的任务。3）.全部分组（Allgrouping）：tuple被复制到bolt的所有任务。这种类型需要谨慎使用。4）.全局分组（Globalgrouping）：全部流都分配到bolt的同一个任务。明确地说，是分配给ID最小的那个task。5）.无分组（Nonegrouping）：你不需要关心流是如何分组。目前，无分组等效于随机分组。但最终，Storm将把无分组的Bolts放到Bolts或Spouts订阅它们的同一线程去执行（如果可能）。6）.直接分组（Directgrouping）：这是一个特别的分组类型。元组生产者决定tuple由哪个元组处理者任务接收。当然还可以实现CustomStreamGroupimg接口来定制自己需要的分组。storm和hadoop的对比来了解storm中的基本概念。HadoopStorm系统角色JobTrackerNimbusTaskTrackerSupervisorChildworker应用名称JobTopology组件接口Mapper/ReducerSpout/Bolt3.Storm应用场景

Storm与其他大数据解决方案的不同之处在于它的处理方式。Hadoop

在本质上是一个批处理系统。数据被引入Hadoop

文件系统(HDFS)并分发到各个节点进行处理。当处理完成时，结果数据返回到HDFS供始发者使用。Storm支持创建拓扑结构来转换没有终点的数据流。不同于Hadoop

作业，这些转换从不停止，它们会持续处理到达的数据。Twitter列举了Storm的三大类应用：1.信息流处理{Streamprocessing}Storm可用来实时处理新数据和更新数据库，兼具容错性和可扩展性。即Storm可以用来处理源源不断流进来的消息，处理之后将结果写入到某个存储中去。2.连续计算{Continuouscomputation}Storm可进行连续查询并把结果即时反馈给客户端。比如把Twitter上的热门话题发送到浏览器中。3.分布式远程程序调用{DistributedRPC}Storm可用来并行处理密集查询。Storm的拓扑结构是一个等待调用信息的分布函数，当它收到一条调用信息后，会对查询进行计算，并返回查询结果。举个例子DistributedRPC可以做并行搜索或者处理大集合的数据。通过配置drpc服务器，将storm的topology发布为drpc服务。客户端程序可以调用drpc服务将数据发送到storm集群中，并接收处理结果的反馈。这种方式需要drpc服务器进行转发，其中drpc服务器底层通过thrift实现。适合的业务场景主要是实时计算。并且扩展性良好，可以增加每个节点的工作worker数量来动态扩展。4.项目架构系统需要满足的要求实时的在

任意大小的数据集上执行各种类型的操作处理架构设计SpeedLayerServingLayerBatchLayer架构设计SpeedLayerServingLayerBatchLayerBatchLayerquery=function(alldata)1、每次查询都在数据集上重新执行一次2、预计算BatchLayer批处理层的主要作用就是进行各种复杂的预计算，预计算的结果称作BatchView。再次查询时直接从BatchView中读取结果数据，为了支持快速的随机的查询需求，需要对BatchView中的数据建立索引。BatchLayerAllDataBatchLayerBatchViewBatchViewBatchViewBatchLayer批处理层的主要职责包括：1、存储不变的、持续增加的海量数据集记录2、预计算各种Query操作对应的BatchViewHDFSMapReduceBatchLayer

批处理层将Query的结果保存到了BatchView中，而Hadoop的MapReduce所对应的BatchView是存储在HDFS上一系列文件，我们无法对这些结果文件进行快速的随机的访问。架构设计SpeedLayerServingLayer

BatchLayerServingLayer

ServingLayer的主要作用就是对

BatchView建立索引，

以支持业务层的快速的随机的访问需求。ServingLayer

分布式数据库是ServingLayer的最佳选择。分布式数据库需要满足的条件：1、支持BatchView的批量导入2、支持对BatchView建立索引和随机读取BulkLoad内建的索引机制ServingLayerServingLayer面临的其他问题：BatchLayer的计算逻辑每天执行一次，所以ServingLayer对应的BatchView每天更新一次，从而导致最新的数据无法实时的更新到ServingLayer架构设计SpeedLayerServingLayerBatchLayerSpeedLayerSpeedLayer的主要目的是弥补系统存在的对ServingLayer延迟更新的缺陷。SpeedLayer会对进入到系统的最新的数据进行各种实时的处理，并产生

RealTimeView。SpeedLayerNewDataSpeedLayerRealTimeViewRealTimeViewRealTimeViewSpeedLayerSpeedLayer使用分布式的、可靠的、实时的计算系统--------------Storm。

RealTimeView使用支持随机读、随机写的分布式数据库HBase。SpoutBoltBoltBoltHBasequery=function(alld