Storm初探

1、Storm;开源分布式实时处理框架,具有高容错性,保证每个信息都会得到处理,可以使用任何编程语言开发应用。

2、Storm集群中真正运行拓扑图的有3个实体:工作进程、线程、任务。每个进程可创建多个线程,每个线程可执行多个任务,任务是真正进行数据处理的实体,spout、bolt就是作为一个或多个任务执行的。

    使用ZeroMQ作为底层消息队列,保证信息能够被快速处理。

3、Storm基本概念:

  有别于Hadoop集群上运行的MR Job,Storm集群上运行的是拓扑topology,而且hadoop的MR Job最终会结束,而topology会永远执行,除非手动kii掉。

      在Storm中有两种节点,控制节点(master node)、工作节点(worker node),在控制节点上运行一个叫Nimbus的后台程序,类似于hadoop的jobTracker,负责分发代码、分配任务、监控状态

      每个工作节点运行一个Supervisor的节点,监听分配的任务,根据需要启动/关闭集成,每个工作进程运行topology的子集,一个运行的topology由运行在很多机器上的很多进程组成。

     Nimbus跟Supervisor之间所有的协调工作是通过Zookeeper来完成的,所有的状态都在Zookeeper上或者本地磁盘上,意味着可以通过kiil -9在杀死进程或者重启。

     

    (1) 一个topology由spouts、bolts组成,通过stream groupings将spouts、bolts连接起来

     

     一个Storm会一直运行直到手动kill掉,Storm会重新运行失败的任务,如果一个机器意外停止会将任务转到其它机器上。

     运行一个topology很简单,将所有代码及依赖的jar包打到同一个jar包,然后执行下面的命令:storm jar all-my-code.jar backtype.storm.MyTopology arg1 arg2,这个命令是运行MyTopology类中的main方法,并且可以传递参数,main方法中定义运行的topology,storm jar 负责连接Nimbus并上传jar。

     (2)streams:信息流

         一个消息流是一个没有边界的tuple序列,这些tuple序列会以分布式的方式进行创建和处理

         每个消息流在定义的时候会分配一个id

        Storm提供的最基本的处理Stream的原语是spout、bolt,可以实现spout、bolt提供的接口来实现自己的业务逻辑

     (3)spouts:消息源

       topology中消息的产生者,消息源会从外部源读取数据并向topology发送消息tuple,spout里最重要的方法就是nextTuple

      (4)bolts:消息处理

       所有消息处理逻辑都放在bolts里,bolts可以做很多事:过滤、聚合、查询数据库等

        使用OutputFieldsDeclarer.declareStream定义stream,使用OutputCollector.emit来选择要发射的stream。

      bolts中主要方法就是execute方法,以tuple作为输入,用OutputCollector来发射tuple,bolts要为每一个处理的tuple消息做OutputCollector.ack处理,以通知storm该tuple处理成功了,再通知这个消息的发送者spouts,

      一般的流程是: bolts处理一个输入tuple,  发射0个或者多个tuple, 然后调用ack通知storm自己已经处理过这个tuple了。storm提供了一个IBasicBolt会自动调用ack。

       (5)streaming groups:信息分组

        定义一个topology时需要定义每个bolt接收什么样的流作为输入,streaming groups就是用来定义一个stream如何将数据分配给bolts上的多个tasks

        Storm中有7种类型的Stream group:

    1>Shuffle Grouping: 随机分组, 随机派发stream里面的tuple,保证每个bolt接收到的tuple数目大致相同。

    2>Fields Grouping:按字段分组, 比如按userid来分组, 具有同样userid的tuple会被分到相同的Bolts里的一个task, 而不同的userid则会被分配到不同的bolts里的task。

    3>All Grouping:广播发送,对于每一个tuple,所有的bolts都会收到。

    4>Global Grouping:全局分组, 这个tuple被分配到storm中的一个bolt的其中一个task。再具体一点就是分配给id值最低的那个task。

    5>Non Grouping:不分组,这个分组的意思是说stream不关心到底谁会收到它的tuple。目前这种分组和Shuffle grouping是一样的效果, 有一点不同的是storm会把这个bolt放到这个bolt的订阅者同一个线程里面去执行。

    6>Direct Grouping: 直接分组, 这是一种比较特别的分组方法,用这种分组意味着消息的发送者指定由消息接收者的哪个task处理这个消息。 只有被声明为Direct Stream的消息流可以声明这种分组方法。而且这种消息tuple必须使用emitDirect方法来发射。消息处理者可以通过TopologyContext来获取处理它的消息的task的id (OutputCollector.emit方法也会返回task的id)。

    7>Local or shuffle grouping:如果目标bolt有一个或者多个task在同一个工作进程中,tuple将会被随机发生给这些tasks。否则,和普通的Shuffle Grouping行为一致。

       (6)reliable:每个topology都有一个消息超时的设置,如果storm在这个超时的时间内检测不到某个tuple树到底有没有执行成功, 那么topology会把这个tuple标记为执行失败,并且过一会儿重新发射这个tuple

           为了利用Storm的可靠性特性,在你发出一个新的tuple以及你完成处理一个tuple的时候你必须要通知storm。这一切是由OutputCollector来完成的。通过emit方法来通知一个新的tuple产生了,通过ack方法通知一个tuple处理完成了。

       (7)tasks:可以调用TopologyBuilder类的setSpout和setBolt来设置并行度(也就是有多少个task)。(task任务个数也称为并行度)

       (8)works:一个topology会在一个或者多个worker里执行,每个worker是一个物理JVM并且执行topology的一部分

       (9)Configuration:配置文件,有两个级别:系统级别、topology级别

           default.yaml里面有所有的默认配置。你可以通过定义个storm.yaml在你的classpath里来覆盖这些默认配置。并且你也可以在代码里面设置一些topology相关的配置信息(使用StormSubmitter) 

原文地址:https://www.cnblogs.com/zhli/p/4819484.html