storm学习笔记.docx

Storm对比Hadoop的批处理，Storm是个实时的、分布式以及具备高容错的计算系统。同Hadoop一样Storm也可以处理大批量的数据，然而Storm在保证高可靠性的前提下还可以让处理进行的更加实时；也就是说，所有的信息都会被处理。Storm同样还具备容错和分布计算这些特性，这就让Storm可以扩展到不同的机器上进行大批量的数据处理。他同样还有以下的这些特性： 易于扩展。对于扩展，你只需要添加机器和改变对应的topology（拓扑）设置。Storm使用Hadoop Zookeeper进行集群协调，这样可以充分的保证大型集群的良好运行。 每条信息的处理都可以得到保证。 Storm集群管理简易。 Storm的容错机能：一旦topology递交，Storm会一直运行它直到topology被废除或者被关闭。而在执行中出现错误时，也会由Storm重新分配任务。 尽管通常使用Java，Storm中的topology可以用任何语言设计。当然为了更好的理解文章，你首先需要安装和设置Storm。需要通过以下几个简单的步骤： 从Storm官方下载Storm安装文件 将bin/directory解压到你的PATH上，并保证bin/storm脚本是可执行的。Storm组件Storm集群主要由一个主节点和一群工作节点（worker node）组成，通过 Zookeeper进行协调。主节点：主节点通常运行一个后台程序 Nimbus，用于响应分布在集群中的节点，分配任务和监测故障。这个很类似于Hadoop中的Job Tracker。工作节点：工作节点同样会运行一个后台程序 Supervisor，用于收听工作指派并基于要求运行工作进程。每个工作节点都是topology中一个子集的实现。而Nimbus和Supervisor之间的协调则通过Zookeeper系统或者集群。ZookeeperZookeeper是完成Supervisor和Nimbus之间协调的服务。而应用程序实现实时的逻辑则被封装进Storm中的“topology”。topology则是一组由Spouts（数据源）和Bolts（数据操作）通过Stream Groupings进行连接的图。下面对出现的术语进行更深刻的解析。Spout：简而言之，Spout从来源处读取数据并放入topology。Spout分成可靠和不可靠两种；当Storm接收失败时，可靠的Spout会对tuple（元组，数据项组成的列表）进行重发；而不可靠的Spout不会考虑接收成功与否只发射一次。而Spout中最主要的方法就是nextTuple（），该方法会发射一个新的tuple到topology，如果没有新tuple发射则会简单的返回。Bolt：Topology中所有的处理都由Bolt完成。Bolt可以完成任何事，比如：连接的过滤、聚合、访问文件/数据库、等等。Bolt从Spout中接收数据并进行处理，如果遇到复杂流的处理也可能将tuple发送给另一个Bolt进行处理。而Bolt中最重要的方法是execute（），以新的tuple作为参数接收。不管是Spout还是Bolt，如果将tuple发射成多个流，这些流都可以通过declareStream（）来声明。Stream Groupings：Stream Grouping定义了一个流在Bolt任务间该如何被切分。这里有Storm提供的6个Stream Grouping类型：1. 随机分组（Shuffle grouping）：随机分发tuple到Bolt的任务，保证每个任务获得相等数量的tuple。2. 字段分组（Fields grouping）：根据指定字段分割数据流，并分组。例如，根据“user-id”字段，相同“user-id”的元组总是分发到同一个任务，不同“user-id”的元组可能分发到不同的任务。3. 全部分组（All grouping）：tuple被复制到bolt的所有任务。这种类型需要谨慎使用。4. 全局分组（Global grouping）：全部流都分配到bolt的同一个任务。明确地说，是分配给ID最小的那个task。5. 无分组（None grouping）：你不需要关心流是如何分组。目前，无分组等效于随机分组。但最终，Storm将把无分组的Bolts放到Bolts或Spouts订阅它们的同一线程去执行（如果可能）。6. 直接分组（Direct grouping）：这是一个特别的分组类型。元组生产者决定tuple由哪个元组处理者任务接收。当然还可以实现CustomStreamGroupimg接口来定制自己需要的分组。概念 一个Topology可以包含一个或多个worker(并行的跑在不同的machine上), 所以worker process就是执行一个topology的子集, 并且worker只能对应于一个topology 一个worker可用包含一个或多个executor, 每个component (spout或bolt)至少对应于一个executor, 所以可以说executor执行一个compenent的子集, 同时一个executor只能对应于一个component Task就是具体的处理逻辑对象, 一个executor线程可以执行一个或多个tasks 但一般默认每个executor只执行一个task, 所以我们往往认为task就是执行线程, 其实不然 task代表最大并发度, 一个component的task数是不会改变的, 但是一个componet的executer数目是会发生变化的 当task数大于executor数时, executor数代表实际并发数 A worker process executes a subset of a topology. A worker process belongs to a specific topology and may run one or more executors for one or more components (spouts or bolts) of this topology. A running topology consists of many such processes running on many machines within a Storm cluster. An executor is a thread that is spawned by a worker process. It may run one or more tasks for the same component (spout or bolt). A task performs the actual data processing each spout or bolt that you implement in your code executes as many tasks across the cluster. The number of tasks for a component is always the same throughout the lifetime of a topology, but the number of executors (threads) for a component can change over time. This means that the following condition holds true: #threads #tasks. By default, the number of tasks is set to be the same as the number of executors, i.e. Storm will run one task per thread. Configuring the parallelism of a topology, 并发度的配置 The following sections give an overview of the various configuration options and how to set them in your code. There is more than one way of setting these options though, and the table lists only some of them. Storm currently has the following order of precedence for configuration settings: defaults.yaml storm.yaml topology-specific configuration internal component-specific configuration executortask要想提高storm的并行度可以从三个方面来改造worker(进程)executor(线程)task(实例)增加work进程，增加executor线程，增加task实例看下面的图：这表示是一个work进程，其实就是一个jvm虚拟机进程，在这个work进程里面有多个executor线程，每个executor线程会运行一个或多个task实例。一个task是最终完成数据处理的实体单元。(默认情况下一个executor运行一个task)worker,executor,task解释1个worker进程执行的是1个topology的子集（注：不会出现1个worker为多个topology服务）。1个worker进程会启动1个或多个executor线程来执行1个topology的component(spout或bolt)。因此，1个运行中的topology就是由集群中多台物理机上的多个worker进程组成的。executor是1个被worker进程启动的单独线程。每个executor只会运行1个topology的1个component(spout或bolt)的task（注：task可以是1个或多个，storm默认是1个component只生成1个task，executor线程里会在每次循环里顺序调用所有task实例）。task是最终运行spout或bolt中代码的单元（注：1个task即为spout或bolt的1个实例，executor线程在执行期间会调用该task的nextTuple或execute方法）。topology启动后，1个component(spout或bolt)的task数目是固定不变的，但该component使用的executor线程数可以动态调整（例如：1个executor线程可以执行该component的1个或多个task实例）。这意味着，对于1个component存在这样的条件：#threadsexecutor(线程)task(实例) 是如何设置的l worker(进程)：这个worker进程数量是在集群启动之前配置好的，在哪配置的呢？是在storm/conf/storm.yaml文件中，参数是supervisor.slots.port，如果我们不在这进行配置的话，这个参数也是有默认值的，在strom-0.9.3的压缩包中的lib目录下，有一个strom-core.jar，打开这个jar文件，在里面有一个defaults.yaml文件中是有一些默认配置的。默认情况下一个storm项目只使用一个work进程，也可以通过代码进行修改，通过config.setNumWorkers(workers)设置。(最好一台机器上的一个topology只使用一个worker,主要原因时减少了worker之间的数据传输)注意：如果worker使用完的话再提交topology就不会执行，因为没有可用的worker，只能处于等待状态，把之前运行的topology停止一个之后这个就会继续执行了，l executor(线程)：默认情况下一个executor运行一个task，可以通过在代码中设置builder.setSpout(id,spout, parallelism_hint);或者builder.setBolt(id,bolt,parallelism_hint);来提高线程数的。l task(实例)：通过boltDeclarer.setNumTasks(num);来设置实例的个数默认情况下，一个supervisor节点会启动4个worker进程。每个worker进程会启动1个executor，每个executor启动1个task。Ok，这几个参数都可以使用一些方法进行增加。下面来举个例子看一下对这些配置修改之后的效果l worker(进程)，通过在代码中设置，可以在ui界面上查看worker的总数，并且还可以在linux服务器上执行jps查看work进程。在代码中设置使用3个worker，查看ui界面，发现workers是3个，executors使用了5个，为什么呢？因为每一个worker默认都会占用一个executor（这个executor会启动一个acker任务），这样就会占用三个，剩下的两个是spout和bolt实例占用了。如果使用5个worker，executor会使用7个，因为worker本身就会占用5个，spout和bolt占用两个。Acker任务默认是每个worker进程启动一个executor线程来执行,，可以在topology中取消acker任务，这样的话就不会多出来一个executor和任务了。代码如下：实际上就是修改一个配置topology.acker.executors这样的话在页面查看就只有两个executor和2个task了。l executor(线程)，在spou