从Storm学习集群管理

简介

Storm是当前最流行的分布式实时计算平台，使用场景是根据Storm定义的接口规范编写一个实时处理流，然后提交到Storm平台处理，Storm平台解析该处理流，使其并行、分布式地在集群中运行，并附带相应的状态监控。本文主要描述Storm的集群管理这块的内容，处理流的相关接口逻辑规范不作涉及。

Storm集群监控管理的目标是管理和监控用户提交的处理流作业（类似于Hadoop监控管理mr-job）。

Storm系统架构

Storm遵循经典的master-slave架构，nimbus节点即为master，supervisor节点即为slave。

nimbus节点负责集群的管理和任务分配工作，supervisor节点负责具体处理流作业的执行。

处理流作业在supervisor节点上面是使用独立的进程process（即图中的worker进程）运行的，进程内部进一步产生执行器（即线程）来执行task。

集群管理基本对象

Storm首先会把处理流解析出若干个task（最小执行单元），并根据并行度来判断启用多少的进程worker-process来运行这些task，随后task会根据一定的规则分配到相应的worker-process。

进程作为基本对象，其好处有：

1. 进程可以很方便地分布在集群各个机器节点，有利于分布式和并行计算。

2. 同一个机器节点内部可以产生多个进程，其相互独立，这样可以充分利用机器节点性能。

3. 进程可以使用cgroup进行资源隔离管理。

进程内部可以产生若干线程来执行若干个task，但这不是分布式系统所必须的，如果进程直接满足，则可以不产生线程来执行task，而是由进程直接执行task，这个完全由系统业务需要和实现逻辑决定。Task作为任务基本执行单元，被纳入Storm集群监控对象。

集群状态监控-zk

nimbus节点作为master，负责整个集群监控管理，但是集群节点运行的状态数据要放到Zookpeer上面。Master的HA也使用了Zookpeer。

Storm在Zookpeer保存的状态数据结构（截取了一部分）：

（参考：http://xumingming.sinaapp.com/466/twitter-Storm-code-analysis-zookeeper-dirs/）

Zookpeer的文件系统式的数据模型很清晰地展现了整个集群里面的运行状态，读写都很方便。

注：个人认为，如果只是保存集群状态，不一定要非要使用Zookpeer，完全可以使用redis这样的第三方来保存和状态轮询，而且master的HA也不一定要使用Zookpeer的leader选举用例，可选方案有很多；但不可以否认的是Zookpeer的数据模型非常的优雅和方便。

Reblance

Work节点的增加、宕机或删除，任务的计算资源扩容或减小都需要对集群里面的基本执行单元（work进程）进行reblance，即弹性计算，这个是集群管理最重要的。Strom集群可以动态增加work进程数量，也可以动态增加work进程内部的task的并发度。本文只陈述work进程这个基本执行单元的弹性扩容。

集群可能存在的动态变化：

1. 一个work节点挂掉，如work-A节点挂掉

2. 添加一个新的work节点，如节点work-C

3. 一个work进程挂掉，如P1进程挂掉

4. 添加一个work进程，如p-New进程，提高任务的计算能力

5. 删除一个work进程挂掉，如p3进程，以减小用户任务的计算资源

以上的变化可以归为两类：添加资源、删除资源。无论是资源的添加和删除，都需要将用户的任务重新分配或调整运行资源。重新分配的策略一般有两种：

1. 重新分配所有的运行资源，

重新分配所有的运行资源，即和用户刚添加任务的时候是一样的分配逻辑，删除所有之前的运行资源（即work进程），然后重新根据任务的资源要求和集群的资源情况进行重新分配。

这种方式逻辑简单，但是显然很粗暴，所有的资源之前的运行状态都全部失效，需要重新加载，在特定的业务场景下面可能存在一些问题。Storm里面采用了这个方式，不过Storm没有采用这个方式，使用了下面说这个分配策略。

2. 保存正在运行的资源不变，只调整变化的资源

这种方式处理起来要麻烦一些，需要考虑当前已经运行的资源情况，在调整变化的资源情况。

ü 新增

新增一个运行资源相对要简单一些，因为变化对现有的运行资源产生影响，只要在集群的资源池里面请求一个新资源加入到任务的资源映射里面就可以了。

ü 删除

删除任务的一个运行资源就要复杂很多，Storm里面可以看到相关的逻辑。说一下我个人的看法：和用户的业务场景有很大关系。

2 如果运行的资源之间的业务完全独立，比如像storm只是做数据流的计算处理，删除一个计算节点，只是计算能力减小，并不会产生业务上面的理解错误。那简单地删除一个运行资源就可以，并不需要做多少额外的判断处理。

2 如果运行的资源是各自负责某个部分的数据处理，那其中一个资源节点的删除旧考虑到其他资源节点的数据分布情况了。一般可以使用类似一致性hash的方法来分配运行资源，尽量保证影响少数的其他运行资源。

Cgroup

进程是集群管理的基本单位，task（进程内部）是与业务计算逻辑有关可以暂不考虑。同一个机器节点内部的进程一般使用Cgroup来进行隔离和管理。Storm也使用了Cgroup进行work进程隔离管理。

Cgroup是control groups的缩写，是Linux内核提供的一种可以限制、记录、隔离进程组（process groups）所使用的物理资源（如：CPU, Memory, IO等）的机制。

Cgroup是使用类似文件系统的模型来进行进程组隔离管理机制。内部有几个子系统控制器来控制进程组。

子系统控制器：

? cpuset 为 cgroup 任务分配各个 CPU 和内存节点

? cpu 调度 CPU 访问（例如，根据相对份额，如图
1 所示，或针对实时进程）

? cpuacct 报告所用的总CPU 时间。

? memory 报告或限制内存使用。

? devices 授予或拒绝对设备的访问权限。

? freezer 挂起或恢复任务。

? net_cls 使用标识符标记传出网络数据包。

? blkio 报告或控制块设备的 I/O 带宽。

通过定义类似如下的层次结构就可以控制进程组：

通过cgroup可以把用户提交的任务进程及任务内部产生的子进程都完全管理和隔离起来。

Docker

像Strom、hadoop这样的专用计算集群，需要用户根据指定的规范和接口来添加任务，然后平台里面自动解析出所需要运行的进程，分布到集群各个机器节点上面运行，进程的运行依赖受系统的种种限制。如果是一个开放式的，通用的分布式系统平台，个人认为可以借用目前火热的轻量级的虚拟化应用容器-docker来实现用户自由提交进程运行容器。

Docker原理是使用cgroup和LXC等linux系统容器机制实现对用户的进程进行更高层级的包装。Docker的初衷也就是将各种应用程序和他们所依赖的运行环境打包成标准的container/image,进而发布到不同的平台上运行。

这样，用户可以定义自己的进程运行容器，这些容器通过docker机制实现标准化，统一化，可以很一致地被集群调度运行。

注：个人对docker机制还不是很了解，继续关注和学习中。

Kubernetes

作为如今火热的docker生态圈中的一员——Kubernetes，是 Google 开源的容器集群管理系统，其提供应用部署、维护、扩展机制等功能，利用Kubernetes能方便地管理跨机器运行容器化的应用，其主要功能如下：

1) 使用Docker对应用程序包装(package)、实例化(instantiate)、运行(run)。

2) 以集群的方式运行、管理跨机器的容器。

3) 解决Docker跨机器容器之间的通讯问题。

4) Kubernetes的自我修复机制使得容器集群总是运行在用户期望的状态。

YARN

说到集群资源管理必须要提到YARN——资源调度框架，现在各大计算平台都可以跑到yarn上面，如hadoop-yarn、storm on yarn、spark on yarn等。

三个角色：RM、AM、Container

? RM：负责全局的资源管理和调度

? AM：一个应用的master，负责应用内的资源申请和管理，与应用逻辑相关

? Container：资源运行单位

YARN的介绍这里不再赘述，网上有很多，这里想说一下，YARN里面的Container，和上面说的docker容器的区别和联系：

Yarn里面的Container也是基于cgroup机制实现用户的进程资源管理和管理，而Docker容器是一个进程运行的标准化的、统一的完整镜像，包括了依赖环境，即隔离管理等机制，也就是说docker容器更加彻底和统一。目前社区也有人在跟进yarn的容器和docker融合的工作，希望后面可以看新的进展。