首页 > 代码库 > 2017.4.5 Strom

2017.4.5 Strom

Strom是分布式实时计算系统,它对于实时计算的意义类似于hadoop对于批处理的意义。与Storm关系密切的语言:核心代码用clojure书写,实用程序用python开发,使用java开发拓扑。

Storm的适用场景

1.流数据处理。Storm可以用来处理源源不断流进来的消息,处理之后将结果写入到某

个存储中去。

2.分布式rpc。由于storm的处理组件是分布式的,而且处理延迟极低,所以可以作为一个通用的分布式rpc框架来使用。当然,其实我们的搜索引擎本身也是一个分布式rpc系统。

 

下图为Apache Storm官网描述:

 

技术分享

在Storm中,先要设计一个用于实时计算的图状结构,我们称之为拓扑(topology)。这个拓扑将会被提交给集群,由集群中的主控节点(master node)分发代码,将任务分配给工作节点(worker node)执行。一个拓扑中包括spout和bolt两种角色,其中spout发送消息,负责将数据流以tuple元组的形式发送出去;而bolt则负责转换这些数据流,在bolt中可以完成计算、过滤等操作,bolt自身也可以随机将数据发送给其他bolt。由spout发射出的tuple是不可变数组,对应着固定的键值对。

技术分享

 

首先我们通过一个storm 和hadoop的对比来了解storm中的基本概念。

技术分享

 

下面我们开始具体讲解Storm的基本概念和它内部的一些实现原理吧。

topology

    Storm集群中有两种节点,一种是控制节点(Nimbus节点),另一种是工作节点(Supervisor节点)。所有Topology任务的 提交必须在Storm客户端节点上进行(需要配置 storm.yaml文件),由Nimbus节点分配给其他Supervisor节点进行处理。 Nimbus节点首先将提交的Topology进行分片,分成一个个的Task,并将Task和Supervisor相关的信息提交到 zookeeper集群上,Supervisor会去zookeeper集群上认领自己的Task,通知自己的Worker进程进行Task的处理。 

    和同样是计算框架的MapReduce相比,MapReduce集群上运行的是Job,而Storm集群上运行的是Topology。但是Job在运行结束之后会自行结束,Topology却只能被手动的kill掉,否则会一直运行下去 

    Storm不处理计算结果的保存,这是应用代码需要负责的事情,如果数据不大,你可以简单地保存在内存里,也可以每次都更新数据库,也可以采用NoSQL存储。这部分事情完全交给用户。 

    数据存储之后的展现,也是你需要自己处理的,storm UI 只提供对topology的监控和统计。 

 

Component

    Storm中,Spout和Bolt都是Component。所以,Storm定义了一个名叫IComponent的总接口 
    全家普如下:绿色部分是我们最常用、比较简单的部分。红色部分是与事务相关的 
技术分享

Spout

    Spout是Stream的消息产生源, Spout组件的实现可以通过继承BaseRichSpout类或者其他Spout类来完成,也可以通过实现IRichSpout接口来实现 
public interface ISpout extends Serializable { 
  void open(Map conf, TopologyContext context, SpoutOutputCollector collector); 
  void close(); 
  void nextTuple(); 
  void ack(Object msgId); 
  void fail(Object msgId); 

    open()方法 -- 初始化方法 
    close() -- 在该spout将要关闭时调用。但是不保证其一定被调用,因为在集群中supervisor节点,可以使用kill -9来杀死worker进程。只有当Storm是在本地模式下运行,如果是发送停止命令,可以保证close的执行 
    ack(Object msgId) -- 成功处理tuple时回调的方法,通常情况下,此方法的实现是将消息队列中的消息移除,防止消息重放 
    fail(Object msgId) -- 处理tuple失败时回调的方法,通常情况下,此方法的实现是将消息放回消息队列中然后在稍后时间里重放 
    nextTuple() -- 这是Spout类中最重要的一个方法。发射一个Tuple到Topology都是通过这个方法来实现的。调用此方法时,storm向spout发出请求, 让spout发出元组(tuple)到输出器(ouput collector)。这种方法应该是非阻塞的,所以spout如果没有元组发出,这个方法应该返回。nextTuple、ack 和fail 都在spout任务的同一个线程中被循环调用。 当没有元组的发射时,应该让nextTuple睡眠一个很短的时间(如一毫秒),以免浪费太多的CPU。 
继承了BaseRichSpout后,不用实现close、 activate、 deactivate、 ack、 fail 和 getComponentConfiguration 方法,只关心最基本核心的部分。 
通常情况下(Shell和事务型的除外),实现一个Spout,可以直接实现接口IRichSpout,如果不想写多余的代码,可以直接继承BaseRichSpout 

Bolt

    Bolt类接收由Spout或者其他上游Bolt类发来的Tuple,对其进行处理。Bolt组件的实现可以通过继承BasicRichBolt类或者IRichBolt接口等来完成 
    prepare方法 -- 此方法和Spout中的open方法类似,在集群中一个worker中的task初始化时调用。 它提供了bolt执行的环境 
    declareOutputFields方法 -- 用于声明当前Bolt发送的Tuple中包含的字段(field),和Spout中类似 
    cleanup方法 -- 同ISpout的close方法,在关闭前调用。同样不保证其一定执行。 
    execute方法 -- 这是Bolt中最关键的一个方法,对于Tuple的处理都可以放到此方法中进行。具体的发送是通过emit方法来完成的。execute接受一个 tuple进行处理,并用prepare方法传入的OutputCollector的ack方法(表示成功)或fail(表示失败)来反馈处理结果。 
    Storm提供了IBasicBolt接口,其目的就是实现该接口的Bolt不用在代码中提供反馈结果了,Storm内部会自动反馈成功。如果你确实要反馈失败,可以抛出FailedException 
    通常情况下,实现一个Bolt,可以实现IRichBolt接口或继承BaseRichBolt,如果不想自己处理结果反馈,可以实现 IBasicBolt接口或继承BaseBasicBolt,它实际上相当于自动实现了collector.emit.ack(inputTuple) 

Topology运行流程

    (1)Storm提交后,会把代码首先存放到Nimbus节点的inbox目录下,之后,会把当前Storm运行的配置生成一个 stormconf.ser文件放到Nimbus节点的stormdist目录中,在此目录中同时还有序列化之后的Topology代码文件 
    (2) 在设定Topology所关联的Spouts和Bolts时,可以同时设置当前Spout和Bolt的executor数目和task数目,默认情况下, 一个Topology的task的总和是和executor的总和一致的。之后,系统根据worker的数目,尽量平均的分配这些task的执行。 worker在哪个supervisor节点上运行是由storm本身决定的 
    (3)任务分配好之后,Nimbus节点会将任务的信息提交到zookeeper集群,同时在zookeeper集群中会有workerbeats节点,这里存储了当前Topology的所有worker进程的心跳信息 
    (4)Supervisor 节点会不断的轮询zookeeper集群,在zookeeper的assignments节点中保存了所有Topology的任务分配信息、代码存储目 录、任务之间的关联关系等,Supervisor通过轮询此节点的内容,来领取自己的任务,启动worker进程运行 
    (5)一个Topology运行之后,就会不断的通过Spouts来发送Stream流,通过Bolts来不断的处理接收到的Stream流,Stream流是无界的。 
    最后一步会不间断的执行,除非手动结束Topology。 

Topology运行方式

    在开始创建项目之前,了解Storm的操作模式(operation modes)是很重要的。 Storm有两种运行方式 
    本地运行的提交方式,例: 
     LocalCluster cluster = new LocalCluster(); 
     cluster.submitTopology(TOPOLOGY_NAME, conf, builder.createTopology()); 
     Thread.sleep(2000); 
     cluster.shutdown(); 
    分布式提交方式,例: 
     StormSubmitter.submitTopology(TOPOLOGY_NAME, conf, builder.createTopology()); 


    需要注意的是,在Storm代码编写完成之后,需要打包成jar包放到Nimbus中运行,打包的时候,不需要把依赖的jar都打迚去,否则如果把依赖的 storm.jar包打进去的话,运行时会出现重复的配置文件错误导致Topology无法运行。因为Topology运行之前,会加载本地的 storm.yaml 配置文件。 
    运行的命令如下: storm jar StormTopology.jar mainclass [args] 

storm守护进程的命令

    Nimbus: storm nimbus 启动nimbus守护进程 
    Supervisor: storm supervisor 启动supervisor守护迚程 
    UI:storm ui 这将启动stormUI的守护进程,为监测storm集群提供一个基于web的用户界面。 
    DRPC: storm drpc 启动DRPC的守护进程 

storm管理命令

    JAR:storm jar topology_jar topology_class [arguments...] 
    jar命令是用于提交一个集群拓扑.它运行指定参数的topology_class中的main()方法,上传topology_jar到nimbus, 由nimbus发布到集群中。一旦提交,storm将激活拓扑并开始处理topology_class 中的main()方法,main()方法负责调用StormSubmitter.submitTopology()方法,并提供一个唯一的拓扑(集群)的 名。如果一个拥有该名称的拓扑已经存在于集群中,jar命令将会失败。常见的做法是在使用命令行参数来指定拓扑名称,以便拓扑在提交的时候被命名。 
    KILL:storm kill topology_name [-w wait_time] 
    杀死一个拓扑,可以使用kill命令。它会以一种安全的方式销毁一个拓扑,首先停用拓扑,在等待拓扑消息的时间段内允许拓扑完成当前的数据流。执行 kill命令时可以通过-w [等待秒数]指定拓扑停用以后的等待时间。也可以在Storm UI 界面上实现同样的功能 
    Deactivate:storm deactivate topology_name 
    停用拓扑时,所有已分发的元组都会得到处理,spouts的nextTuple方法将不会被调用。也可以在Storm UI 界面上实现同样的功能 
    Activate:storm activate topology_name 
    启动一个停用的拓扑。也可以在Storm UI 界面上实现同样的功能 
    Rebalance:storm rebalance topology_name [-w wait_time] [-n worker_count] [-e component_name=executer_count]... 
    rebalance使你重新分配集群任务。这是个很强大的命令。比如,你向一个运行中的集群增加了节点。rebalance命令将会停用拓扑,然后在相应超时时间之后重分配worker,并重启拓扑 
例:storm rebalance wordcount-topology -w 15 -n 5 -e sentence-spout=4 -e split-bolt=8 
    还有其他管理命令,如:Remoteconfvalue、REPL、Classpath等 

新建storm项目注意事项

    为了开发storm项目,你的classpath里面需要有storm的jar包。最推荐的方式是使用Maven,不使用maven的话你可以手动把storm发行版里面的所有的jar包添加到classpath 

    storm-starter项目使用Leiningen作为build和依赖管理工具,你可以下载这个脚本 (https://raw.githubusercontent.com/technomancy/leiningen/stable/bin /lein)来安装Leiningen, 把它加入到你的PATH, 使它可执行。要拉取storm的所有依赖包,简单地在项目的根目录执行 lein deps 就可以了

    使用Storm的公司有:Twitter,雅虎,Spotify还有The Weather Channel等。

    说到微批处理,如果你必须有状态的计算,恰好一次的递送,并且不介意高延迟的话,那么可以考虑Spark Streaming,特别如果你还计划图形操作、机器学习或者访问SQL的话,Apache Spark的stack允许你将一些library与数据流相结合(Spark SQL,Mllib,GraphX),它们会提供便捷的一体化编程模型。尤其是数据流算法(例如:K均值流媒体)允许Spark实时决策的促进。

 

Storm与传统关系型数据库比较     

1.传统关系型数据库是先存后计算,而storm则是先算后存,甚至不存 
2.传统关系型数据库很难部署实时计算,只能部署定时任务统计分析窗口数据 
3.关系型数据库重视事务,并发控制,相对来说Storm比较简陋 
4.Storm,Hadoop,Spark等是流行的大数据方案 

 

2017.4.5 Strom