storm从入门到放弃(一)，storm介绍
背景:目前就职于国内最大的IT咨询公司，恰巧又是毕业季，所在部门招了20多个应届毕业生，本人要跟部门新人进行为期一个月的大数据入职培训，特此将整理的文档分享出来。 原文和作者一起讨论:http://www.cnblogs.com/intsmaze/p/7274361.html 微信：intsmaze 避免微信回复重复咨询问题，技术咨询请博客留言。 Storm是一个开源的分布式实时计算系统，可以简单、可靠的处理大量的数据流。Storm支持水平扩展，具有高容错性，保证每个消息都会得到处理。 Storm核心组件 Nimbus：负责资源分配和任务调度，Nimbus对任务的分配信息会落到zookeeper上面的目录下。 Supervisor：负责去zookeeper上的指定目录接受nimbus分配的任务，启动和停止属于自己管理的worker进程。（它是当前物理机器上的管理者）--通过配置文件设置当前supervisor上启动多少个worker。 Worker：运行具体处理组件逻辑的进程。Worker运行的任务类型只有两种，一种是Spout任务，一种是Bolt任务。 Task：worker中每一个spout/bolt的线程称为一个task.在storm0.8之后，task不再与物理线程对应，不同spout/bolt的task可能会共享一个物理线程，该线程称为executor。 Storm一些概念 Topologies ： 拓扑，也俗称一个任务。（可以理解为一个storm集群） Spouts ： 拓扑的消息源。 Bolts ： 拓扑的处理逻辑单元。（一个Bolt类会在集群里面很多机器上并发执行） （Spouts，Bolts可以理解为storm中的两个组件） tuple：消息元组（是在Spouts，Bolts中传递数据的一种封装的格式） Streams ： 流 Stream groupings ：流的分组策略 Tasks ： 任务处理单元 Executor :工作线程 Workers ：工作进程 Configuration ： topology的配置 Storm中的Workers 一个topology可能会在一个或者多个worker（工作进程）里面执行；一个进程里面会启动多个Executor :工作线程。 每个worker是一个物理JVM并且执行整个topology的一部分；在一个物理节点上可以运行一个或多个独立的JVM 进程。一个Topology可以包含一个或多个worker(并行的跑在不同的物理机上), 所以worker process就是执行一个topology的子集, 并且worker只能对应于一个topology。比如，对于并行度是300的topology来说，如果我们使用50个工作进程来执行，那么每个工作进程会处理其中的6个tasks； Storm会尽量均匀的工作分配给所有的worker；一个Executor:工作线程里面可以运行多个相同的task实例。 Storm中的Tasks 每一个spout和bolt会被当作很多task在整个集群里执行；每一个executor对应到一个线程，在这个线程上运行多个task；stream grouping则是定义怎么从一堆task发射tuple到另外一堆task；可以调用TopologyBuilder类的setSpout和setBolt来设置并行度（也就是有多少个task） Executors (threads) 在一个worker JVM进程中运行着多个Java线程。一个executor线程可以执行一个或多个tasks。但一般默认每个executor只执行一个task。（开发中也不建议一个executor里面执行多个task.）一个worker可以包含一个或多个executor, 每个component (spout或bolt)至少对应于一个executor, 所以可以说executor执行一个compenent的子集, 同时一个executor只能对应于一个component。 Storm中的Stream 消息流stream是storm里的关键抽象；一个消息流是一个没有边界的tuple序列， 而这些tuple序列会以一种分布式的方式并行地创建和处理；通过对stream中tuple序列中每个字段命名来定义stream。 在默认的情况下，tuple的字段类型可以是：integer，long，short， byte，string，double，float，boolean和byte array；可以自定义类型（只要实现相应的序列化器）。 数据从一个节点传到另一个节点，数据是要被序列化的，但在storm中，数据序列化之前，消息必须按照一定的格式传递，这个格式就是一个一个的消息元组。消息元组是源源不断的发送的，这个元组就类似一个list，里面有若干个字段。 Storm编程模型 有向无环图 public class RandomSentenceSpout extends BaseRichSpout { public void nextTuple() { collector.emit(new Values("+ - * % /")); Utils.sleep(50000); } ...... } public class SplitSentenceBolt extends BaseBasicBolt { public void execute(Tuple input, BasicOutputCollector collector) { String sentence = (String)input.getValueByField("intsmaze"); System.out.println(Thread.currentThread().getId()+" "+sentence); } ...... } public class TwoBolt extends BaseBasicBolt { public void execute(Tuple input, BasicOutputCollector collector) { String sentence = (String)input.getValueByField("intsmaze"); System.out.println(Thread.currentThread().getId()+" "+sentence); } ...... } public class WordCountTopologyMain { public static void main(String[] args) throws Exception { TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("spout1", new RandomSentenceSpout(),1); builder.setBolt("two", new TwoBolt(),1).shuffleGrouping("spout1"); builder.setBolt("split1", new SplitSentenceBolt(),2).shuffleGrouping("spout1"); Config conf = new Config(); conf.setDebug(false); conf.setMaxTaskParallelism(3); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("word-count", conf, builder.createTopology()); } } } 可以发现spout每隔一段时间间隔发一份数据，这份数据会被两个bolt同时接收，而不是说这次A bolt接收下次B bolt接收。 同一个bolt业务逻辑如果设置了并行度，他们才会根据分组策略依次接收上游发来的消息。 ----------------84 + - * % / 这个是tow bolt接收 ----------------78 + - * % / 这个是split1 bolt 中78线程接收的 ----------------80 + - * % / 这个是split1 bolt中线程80接收的。 ----------------84 + - * % / ----------------78 + - * % / ----------------84 + - * % / Storm7种stream grouping Shuffle Grouping: 随机分组,随机派发stream里面的tuple，保证每个bolt接收到的tuple数目大致相同。 Fields Grouping：按字段分组，比如按userid来分组，具有同样userid的tuple会被分到相同的Bolts里的一个task，而不同的userid则会被分配到不同的bolts里的task。 All Grouping：广播发送，对于每一个tuple，所有的bolts都会收到。 Global Grouping：全局分组， 这个tuple被分配到storm中的一个bolt的其中一个task。再具体一点就是分配给id值最低的那个task。 Non Grouping：不分组，这stream grouping个分组的意思是说stream不关心到底谁会收到它的tuple。目前这种分组和Shuffle grouping是一样的效果，有一点不同的是storm会把这个bolt放到这个bolt的订阅者同一个线程里面去执行。 Direct Grouping： 直接分组， 这是一种比较特别的分组方法，用这种分组意味着消息的发送者指定由消息接收者的哪个task处理这个消息。只有被声明为Direct Stream的消息流可以声明这种分组方法。而且这种消息tuple必须使用emitDirect方法来发射。消息处理者可以通过TopologyContext来获取处理它的消息的task的id （OutputCollector.emit方法也会返回task的id）。 Local or shuffle grouping：如果目标bolt有一个或者多个task在同一个工作进程中，tuple将会被随机发生给这些tasks。否则，和普通的Shuffle Grouping行为一致。 conf.setNumWorkers(4) 表示设置了4个worker来执行整个topology的所有组件 builder.setBolt("boltA-intsmaze",new BoltA(), 4) ---->指明 boltA组件的线程数excutors总共有4个 builder.setBolt("boltB-intsmaze",new BoltB(), 4) ---->指明 boltB组件的线程数excutors总共有4个 builder.setSpout("randomSpout-intsmaze",new RandomSpout(), 2) ---->指明randomSpout组件的线程数excutors总共有2个 -----意味着整个topology中执行所有组件的总线程数为4+4+2=10个 ----worker数量是4个，有可能会出现这样的负载情况，worker-1有2个线程，worker-2有2个线程，worker-3有3个线程，worker-4有3个线程 如果指定某个组件的具体task并发实例数 builder.setSpout("randomspout-intsmaze", new RandomWordSpout(), 4).setNumTasks(8); ----意味着对于这个组件的执行线程excutor来说，一个excutor将执行8/4=2个task，默认情况一个线程执行一个task. 作者： intsmaze(刘洋) 出处： http://www.cnblogs.com/intsmaze/ 老铁，你的--->推荐，--->关注,--->评论--->是我继续写作的动力。 微信公众号号：Apache技术研究院 由于博主能力有限，文中可能存在描述不正确，欢迎指正、补充！ 本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。
