Storm-源码分析汇总

EvenScheduler同DefaultScheduler一样,同样实现了IScheduler接口, 由下面代码可以看出: (ns backtype.storm.scheduler.EvenScheduler (:use [backtype.storm util log config]) (:require [clojure.set :as set]) (:import [backtype.storm.scheduler IScheduler Topologies Cluster Topolo

nimbus-datastorm-core/backtype/storm/nimbus.clj (defn nimbus-data [conf inimbus] (let [forced-scheduler (.getForcedScheduler inimbus)] {:conf conf :inimbus inimbus ; INimbus实现类, standalone-nimbus的返回值 :submitted-count (atom 0) ; 已经提交的计算拓扑的数量, 初始值为原子值0

在Storm中IBasicBolt的主要作用是为用户提供一种更为简单的Bolt编写方式,更为简单体现在Storm框架本身帮你处理了所发出消息的Ack.Fail和Anchor操作,而这部分操作是由执行器BasicBoltExecutor 实现的. 下面我们看一下BasicBoltExecutor的源码: /** * BasicBoltExecutor实现了IRichBolt接口 * 在该类中持有一个IBasicBolt成员变量用于调用转发 * 说明: * 该类是基于装饰模式实现的. */ publ

在"storm源码分析之topology提交过程"一文最后,submitTopologyWithOpts函数调用了mk-assignments函数.该函数的主要功能就是进行topology的任务分配(task assignment).mk-assignments函数定义如下: ;; get existing assignment (just the executor->node+port map) -> default to {};; filter out ones whi

Scheduler作为Storm的调度器,负责为Topology分配可用资源. Storm提供了IScheduler接口,用户可以通过实现该接口来自定义Scheduler. 其定义如下: public interface IScheduler { //接收当前Nimbus的Storm配置作为参数,进行一些初始化工作 void prepare(Map conf); /** * 真正进行任务分配的方法,在Nimbus进行任务分配的时候会调用该方法. * 参数为topologies.cluster:前

本文主要是解析SpoutOutputCollector源码,顺便分析该类中所涉及的设计模式–代理模式. 首先介绍一下Spout输出收集器接口–ISpoutOutputCollector,该接口主要声明了以下3个抽象方法用来约束ISpoutOutputCollector的实现类.接口定义与方法说明如下: /** * ISpoutOutputCollector:Spout输出收集器接口 */ public interface ISpoutOutputCollector { /** * 改方法用来向外

storm集群上运行的是一个个topology,一个topology是spouts和bolts组成的图.当我们开发完topology程序后将其打成jar包,然后在shell中执行storm jar xxxxxx.jar xxxxxxxClass就可以将jar包上传到storm集群的nimbus上,并执行topology.本文主要分析下topology的jar包是如何上传到nimbus上的.首先我们从storm的jar命令入手,jar命令的实现位于storm根目录的bin/storm文件里.定义如

Transactionalspouts Trident是以小批量(batch)的形式在处理tuple,并且每一批都会分配一个唯一的transaction id.不同spout的特性不同,一个transactionalspout会有如下这些特性: 1.有着同样txid的batch一定是一样的.当重播一个txid对应的batch时,一定会重播和之前对应txid的batch中同样的tuples. 2.各个batch之间是没有交集的.每个tuple只能属于一个batch 3.每一个tuple都属于一个b

你已经看到实现有且只有一次被执行的语义时的复杂性.Trident这样做的好处把所有容错想过的逻辑都放在了State里面 -- 作为一个用户,你并不需要自己去处理复杂的txid,存储多余的信息到数据库中,或者是任何其他类似的事情.你只需要写如下这样简单的code:   TridentTopology topology = new  TridentTopology();          TridentState wordCounts =         topology.newStream("sp

nimbus是storm集群的"控制器",是storm集群的重要组成部分.我们可以通用执行bin/storm nimbus >/dev/null 2>&1 &来启动nimbus.bin/storm是一个python脚本,在这个脚本中定义了一个nimbus函数: nimbus函数 def nimbus(klass="backtype.storm.daemon.nimbus"):    """Syntax: [s