基于Twitter Storm的数据实时分析处理工具研究.doc

1、1基于 Twitter Storm 的数据实时分析处理工具研究【摘要】过去的十年是数据处理变革的十年，MapReduce，Hadoop以及一些相关的技术使得我们能处理的数据量比以前要大得多得多。但是这些数据处理技术都不是实时的系统，它们设计的目的也不是为了实时计算。没有什么办法可以简单地把 hadoop 变成一个实时计算系统。然而大规模的实时数据处理已经越来越成为一种业务需求了，而缺少一个“实时版本的 hadoop”已经成为数据处理整个生态系统的一个巨大缺失。twitter storm 的出现弥补了 hadoop 在实时处理方面的不足，本文就twitter storm 在实时数据计算方面的优点

2、和架构实现进行研究。 【关键词】twitter storm 实时计算 实时数据处理 一、Twitter Storm 的优点 Storm 出现之前，你可能需要自己手动维护一个由消息队列和消息处理者所组成的实时处理网络，消息处理者从消息队列取出一个消息进行处理，更新数据库，发送消息给其它队列，等等等等。不幸的是，这种方式有以下几个缺陷： （1）单调乏味：你花费了绝大部分开发时间去配置把消息发送到哪里，部署消息处理者，部署中间消息节点 你的大部分时间花在设计，配置这个数据处理框架上，而你真正关心的消息处理逻辑在你的代码里面占的比例很少。 （2）脆弱：不够健壮，你要自己写代码保证所有的消息处理者和消2

3、息队列正常运行。 （3）伸缩性差：当一个消息处理者的消息量达到阀值，你需要对这些数据进行分流，你需要配置这些新的处理者以让他们处理分流的消息。虽然对于一个大量消息处理系统来说，分解到最后就是消息队列和消息处理者的组合，而消息处理无疑是实时计算的基础。那么现在问题就是：怎样去做才能不丢失数据，可以很好的扩展到更大的消息量并且非常容易操作呢？ Storm 定义了一批实时计算的原语。如同 hadoop 大大简化了并行批量数据处理，storm 的这些原语大大简化了并行实时数据处理。storm 的一些关键特性如下： （1）适用场景广泛： storm 可以用来处理消息和更新数据库（消息流处理） ， 对一个

4、数据量进行持续的查询并返回客户端（持续计算） ， 对一个耗资源的查询作实时并行化的处理（分布式方法调用） ，storm 的这些基础原语可以满足大量的场景。 （2）可伸缩性高：Storm 的可伸缩性可以让 storm 每秒可以处理的消息量达到很高。为了扩展一个实时计算任务，你所需要做的就是加机器并且提高这个计算任务的并行度设置（parallelism setting） 。作为Storm 可伸缩性的一个例证， 一个 Storm 应用在一个 10 个节点的集群上每秒处理 1000000 个消息，包括每秒一百多次的数据库调用。Storm 使用ZooKeeper 来协调集群内的各种配置使得 Storm

5、的集群可以很容易的扩展很大。 3（3）保证无数据丢失： 实时系统必须保证所有的数据被成功的处理。 那些会丢失数据的系统的适用场景非常窄， 而 storm 保证每一条消息都会被处理， 这一点和 S4 相比有巨大的反差。 （4）异常健壮：不像 Hadoop出了名的难管理，storm 集群非常容易管理。容易管理是 storm 的设计目标之一。 （5）容错性好：如果在消息处理过程中出了一些异常，storm 会重新安排这个出问题的处理逻辑。storm 保证一个处理逻辑永远运行，除非你显式杀掉这个处理逻辑。 （6）语言无关性：健壮性和可伸缩性不应该局限于一个平台。Storm 的 topology 和消息处

6、理组件可以用任何语言来定义，这一点使得任何人都可以使用 storm。 二、Twitter Storm 的关键概念 （1）计算拓补（Topologies） ：一个实时计算应用程序的逻辑在storm 里面被封装到 topology 对象里面，我把它叫做计算拓补。Storm里面的 topology 相当于 Hadoop 里面的一个 MapReduce Job，它们的关键区别是：一个 MapReduce Job 最终总是会结束的，然而一个 storm 的topoloy 会一直运行，除非你显式的杀死它。 一个 Topology 是 Spouts和 Bolts 组成的图状结构， 而链接 Spouts 和

7、Bolts 的则是 Stream groupings。 （2）消息流（Streams）：消息流是 storm 里面的最关键的抽象。一个消息流是一个没有边界的 tuple 序列， 而这些 tuples 会被以一种分布式的方式并行地创建和处理。对消息流的定义主要是对消息流里面4的 tuple 的定义，我们会给 tuple 里的每个字段一个名字。并且不同tuple 的对应字段的类型必须一样。也就是说： 两个 tuple 的第一个字段的类型必须一样，第二个字段的类型必须一样，但是第一个字段和第二个字段可以有不同的类型。在默认的情况下，tuple 的字段类型可以是：integer，long，short，

8、byte，string，double，float，boolean 和byte array。你还可以自定义类型，只要你实现对应的序列化器。 每个消息流在定义的时候会被分配给一个 id，因为单向消息流是那么的普遍，OutputFieldsDeclarer 定义了一些方法让你可以定义一个stream 而不用指定这个 id。在这种情况下这个 stream 会有个默认的id：1。 （3）消息源（Spout）：消息源 Spouts 是 storm 里面一个 topology里面的消息生产者。一般来说消息源会从一个外部源读取数据并且向topology 里面发出消息：tuple。消息源 Spouts 可以是可

9、靠的也可以是不可靠的。一个可靠的消息源可以重新发射一个 tuple 如果这个 tuple没有被 storm 成功的处理，但是一个不可靠的消息源 Spouts 一旦发出一个 tuple 就把它彻底忘了，也就不可能再发了。 消息源可以发射多条消息流 stream。要达到这样的效果，使用OutFieldsDeclarer.declareStream 来定义多个 stream，然后使用poutOutputCollector 来发射指定的 sream。Spout 类里面最重要的方法是 nextTuple 要么发射一个新的 tuple 到 topology 里面或者简单的返回如果已经没有新的 tuple

10、了。要注意的是 nextTuple 方法不能 block Spout 的实现， 因为 storm 在同一个线程上面调用所有消息源 Spout 的5方法。 另外两个比较重要的 Spout 方法是 ack 和 fail。storm 在检测到一个 tuple 被整个 topology 成功处理的时候调用 ack， 否则调用fail。storm 只对可靠的 spout 调用 ack 和 fail。 （4）消息处理者：Bolts 所有的消息处理逻辑被封装在 bolts 里面。Bolts 可以做很多事情：过滤，聚合，查询数据库等。Bolts 也可以简单的做消息流的传递。复杂的消息流处理往往需要很多步骤，从

11、而也就需要经过很多 Bolts。比如算出一堆图片里面被转发最多的图片就至少需要两步：第一步算出每个图片的转发数量。第二步找出转发最多的前 10 个图片。 （如果要把这个过程做得更具有扩展性那么可能需要更多的步骤） 。三、Twitter Storm 的设计思想 在 Storm 中也有对于流 stream 的抽象，流是一个不间断的无界的连续 tuple，注意 Storm 在建模事件流时，把流中的事件抽象为 tuple 即元组。 Storm 认为每个 stream 都有一个 stream 源，也就是原始元组的源头，所以它将这个源头抽象为 spout，spout 可能是连接 twitter api 并

12、不断发出 tweets，也可能是从某个队列中不断读取队列元素并装配为 tuple发射。 有了源头即 spout 也就是有了 stream，那么该如何处理 stream 内的tuple 呢，同样的思想 twitter 将流的中间状态转换抽象为 Bolt，bolt可以消费任意数量的输入流，只要将流方向导向该 bolt，同时它也可以6发送新的流给其他 bolt 使用，这样一来，只要打开特定的 spout（管口）再将 spout 中流出的 tuple 导向特定的 bolt，又 bolt 对导入的流做处理后再导向其他 bolt 或者目的地。 我们可以认为 spout 就是一个一个的水龙头，并且每个水龙头

13、里流出的水是不同的，我们想拿到哪种水就拧开哪个水龙头，然后使用管道将水龙头的水导向到一个水处理器（bolt） ，水处理器处理后再使用管道导向另一个处理器或者存入容器中。 为了增大水处理效率，我们很自然就想到在同个水源处接上多个水龙头并使用多个水处理器，这样就可以提高效率。 对应上文的介绍，我们可以很容易的理解，就像一张有向无环图，Storm 将这个图抽象为 Topology 即拓扑。拓扑是 storm 中最高层次的一个抽象概念，它可以被提交到 storm 集群执行，一个拓扑就是一个流转换图，图中每个节点是一个 spout 或者 bolt，图中的边表示 bolt 订阅了哪些流，当 spout 或

14、者 bolt 发送元组到流时，它就发送元组到每个订阅了该流的 bolt（这就意味着不需要我们 手工拉管道，只要预先订阅，spout 就会将流发到适当 bolt 上） 。 为了做实时计算，我们需要设计一个拓扑图，并实现其中的 Bolt 处理细节，Storm 中拓扑定义仅仅是一些 Thrift 结构体（请 google 一下Thrift） ，这样一来我们就可以使用其他语言来创建和提交拓扑。 四、Twitter Storm 的架构实现 Storm 集群表面类似 Hadoop 集群。但在 Hadoop 上你运行的是”MapReduce jobs”，在 Storm 上你运行的是”topologies”

15、。 ”Jobs”和”7topologies”是大不同的，一个关键不同是一个 MapReduce 的 Job 最终会结束，而一个 topology 永远处理消息（或直到你 kill 它） 。 Storm 集群有两种节点：控制（master）节点和工作者（worker）节点。控制节点运行一个称之为”nimbus”的后台程序，它类似于 Haddop的”JobTracker” 。Nimbus 负责在集群范围内分发代码、为 worker 分配任务和故障监测。每个工作者节点运行一个称之”Supervisor”的后台程序。Supervisor 监听分配给它所在机器的工作，基于 Nimbus 分配给它的事 情

16、来决定启动或停止工作者进程。每个工作者进程执行一个topology 的子集（也就是一个子拓扑结构） ；一个运行中的 topology 由许多跨多个机器 的工作者进程组成。 一个 Zookeeper 集群负责 Nimbus 和多个 Supervisor 之间的所有协调工作（一个完整的拓扑可能被分为多个子拓扑并由多个 supervisor 完成） 。 此外，Nimbus 后台程序和 Supervisor 后台程序都是快速失败（fail-fast）和无状态的；所有状态维持在 Zookeeper 或本地磁 盘。这意味着你可以 kill -9 杀掉 nimbus 进程和 supervisor 进程，然后重启，它们将恢复状态并继续工作，就像什么也没发生。这种设计使 storm极其稳定。这种设计 中 Master 并没有直接和 worker 通信，而是借助一个中介 Zookeeper，这样一来可以分离 master 和 worker 的依赖，将状态信息存放 在 zookeeper 集群内以快速回复任何失败的一方。