数据处理神器storm的理解与思考 ——让你的数据化作行云流水

大数据之殇

要问storm是什么?简单答复就是：storm对于实时计算的相当于hadoop对于批处理。两者代表的对大数据处理的两种不同方式与态度，即hadoop代表的批处理方式,与storm为代表的流式计算。
先不扯流式计算是个什么鬼。如果说到大数据分析，大家首先直观就会想到hadoop的批处理方式。不管hadoop的图标上面的大象画得有多萌，出现在大家脑中的画面里的，肯定都会有一个庞然大物，好似几个大力巨神在移山搬海。即然是大数据，你自然需要一个能容纳海量数据的存储，为了兼顾效率与可靠，hdfs、hbase这样的工具应运而生。MapReduce的计算框架在帮你降低编程难度的同时，通过以计算能力去求找数据的方式，减少了数据传输的量，但是仍会有大规模的数据需要集中传输，占用大量带宽。由于批处理是对数据的大量数据的集中处理，强大的计算能力必不可缺，甚至有些场景，巨大的内存使用量也是让你望还却步的。可见批处理的处理思想虽然也有很多分布式的概念在，但总体感觉还是在是以大制大。你量大，我就力气要大。这就导致大存储，大带宽，大计算能力，大内存的需求。所以对很多人来说，这位移山大神不是你请得起的。

更糟糕的一点是，就算你请下了这位大神，你的这些硬件资源大多数情况下的浪费，很多MR的job, 我们都会定时在凌晨处理的，为了避免大量网络带宽占有对其他类数的影响，大多数这些资源很可能是闲置的，积累的数据是不动的。
此外，由于批处理的特点，数据一般都是先积累再做离线处理的。难免数据不是实时的最新的。不少场景下还需要另一套系统来补实时的缺。典型的场景还是搜索引擎中的全量与增量的情况。
那么，作为流式处理的代表的storm，又是如何用另一种方式达到四两拨千斤，举重若轻的效果？ 与批处理先积累再一举歼灭的方式不同。流式计算处理数据，将汪洋分作涓涓细流，随波而流，将山岳化成沙石，乘风而动，再经流转沉淀，细流汇流入海，沙石累土成山，以生生不止之势大浪淘沙，最终积淀成我们的数据金矿。差不多够了，不装逼了，后面几节具体说明其思想与原理。
本文并不讲解storm的用法与内部的角色概念，主要是抽取一些让我兴奋的设计思想供大家参考。注意，虽然本文为了作文效果，把storm吹得各种NB，在很多情况下，流式计算是难以替代批处理，否则你发会现过程艰辛，代价沉重。在下一篇文章中会分析从批处理到流式计算实践中遇到的问题。

 

让数据飞——storm的编程模式

在介绍strom时，会提到很多优点，比如实时、可扩展（感觉对单个topo的扩展能力还是有限）、鲁棒性、数据的可靠性、容错机制等等。对我而言，它最大的优点是提供了安全简易的一种编程模式，让我们的代码可以更加专注业务。
众所周知，java在处理各种多线程情况下的能力是非常强大的，提供了各种各样的便于使用的类库，也正因为此，java的服务器后台以及中间件相关的优秀产品层出不重。但是对于开发而言，由于多线程的竞争而导致的多种安全问题就让人比较头疼。而且在很多数据处理的应用场景下，为了提处理效率，你必然会大量的引入多线程的处理，此时烦恼才刚刚开始。
试想，一个数据处理任务，经常可能是I/O占用型的，因为，你总要有数据的传输与输出的地方，有可能是网络传输，也有可能是磁盘读写，如果单线程的模式下，你会发现你的进程的cpu时间大部份都被I/O读写的等待所占用，所以你就需要多线程的并发方式，以提高cpu的使用，增加你的处理速度。新的问题又产生了，就是如何利用多线程安全地去处理这些数据。很有时候这个问题就可以转换成为“生产者-消费者”这种经典模式来解决。因为我们要处理的就是数据从来源到分发的过程。在翻看与编写各种后台数据处理代码，或者开源中间件框架源码时，就会发觉 master-worker这种并发模型（个人认为后台开发者必须掌握，不了解的请自行扫盲）俯拾皆是。问题不止于此，在消息分发的过程中你很有可能还要考虑消息分发方法，是否需要保证消息序性 （关于有序性，我在《闲扯kafka mq》 中的 并行与有序的矛盾 一节中亦有提到）。再回首看我们写的数据分析的代码，原来我们的大部份精力与代码，都是用于处理上述“生产者-消费者”问题的重复工作上，当数据处理过程包括多个环节的时候，代码中就很有可能多次嵌入这种master-worker的模式。分布式环境下，有时还不得不自行引入并维护外部的消息处理系统，增加你的系统的复杂性。
可以说，storm让开发者从上述烦恼中解放出来，让你的编程变得简单，并且你的代码部份只需要关注具体的数据处理逻辑。总的来说，我们只需要实现两种类，一个是spout（数据源的流入），一个是bolt（数据处理节点）。只要根据的具体逻辑实现指定接口便可。更重要的是，你完全不用去思考线程安全的问题，因为你编写的bolt与spout都是在各自独立的单线程中运行的，除了一些配置信息，基乎没有任何数据共享。如果一旦你在你的bolt或者spout中处理逻辑中开始自行新启包含复杂处理的线程，又开始在多线程的竞争中迷失纠结时，可以恭喜您了，您的任务不适合storm，或者您的storm的使用方式不当。
当然，由于storm是一种分布式系统，一开始就引入了轻量的zeroMq作为消息队列。事实上，最新的0.9.3版本默认使用的是netty，帮你解决了不同jvm之间消息分发的问题。程序员们又可以少操一份心了。
此外storm的多种分组策略（grouping）也帮开发者解决了分发方式与有序性的问题。

 

流水线的艺术

毋庸置疑，Storm的设计是模访hadoop的，hadoop的MapReduce同样也是一种让你专注业务的编程模式。但是，有一点很重要的区别， Hadoop的MapReduce方式是数据在处理流程中是要落地的，而storm在处理流程中你的数据永远只是个过客，基本不做逗流，一直在飞。
尽管MapReduce在Map阶段可以使用本地数据优化的方式，减少数据传输量，但在进入reducer之前，数据仍会被写入磁盘，然后做shuffle排序，再发送。每次查看磁盘读写若者带宽占用的相关的报表时，总会看到在特定时间点的一个个刺尖。特别是宽带的占用，也让开发者吃了不少的苦头，因为磁盘与cpu的短时占用，虽说危险，总归是在一台机器上发生的，不一定会影响到其他的服务。而带宽的占用，影响的就不仅是当前的数据分析任务，同一机房的所有服务都有可能被影响，导致各种超时，搞瘫一片的服务，心醉不已。因此在很多场景下，我们更渴望一种带宽占用更为平滑的处理方式。流式计算就能做到这样的效果。
最初听到流式计算这个名词，感觉相当唬人。说白了，就跟现实中工厂里的流水线差不多。流式计算topo图上的每个处理节点，就相当于每条流水线上的一个工人，重复单一地作着各自的工作。这就保证了每个bolt节点的处理工作尽可能地简单单一，不用去关心太多，轻量级地在一个线程内无脑跑就是了。
流水线式的生产方法是一个伟大的发明，极大地提高了工作的效率。想想富土康，各条流水线上劳作的工人，你又会认为它一个残酷的发明，不光是重复工作的艰辛，工作的无聊单调使人发狂。这是对人性的摧残，如若换成机器的呢，正是这样的工作让电脑cpu跑得飞机，说不定这样的任务会让它们兴奋不已。
    某天你一早醒来，发现你家pc的cpu有了灵性，能开口说话了。你上去凑凑近乎，搭个讪，
    “你平时都有什么兴趣爱好啊？”
    “死循环。”
     “……”
Storm的流式计算所经过的数据总是来去匆匆。 正常情况下，你并不需要为你的woker分配大太的内存，因为每条数据都不会作太久的逗流，如果你的业务在中间某个bolt节点上做不断累积数据，占用太多内存或者磁盘，照样要恭喜你，你的任务还是有太多批处理的特点，不太适合storm的流式计算。
可能有人会怀疑，storm把原来放在一起处理的工作打断，分成多个环节，交给分布在不同机器上的节点去做计算，虽说每个节点的任务简化了，但是增加了网络传输的成本，原来在一台机器上不需要就多余传输的任务，你现在需要不断在多台机器上做网络传输，这样不是很多cpu时间浪费在网络I/O了吗？你的工作效率能提高吗？
我简单地做一下测试，跑了一下这前完成的一个topology了，从kafka发往storm处理，在两台机器上跑storm，能跑出2.5w条数据每秒的tps出来，每个worker的cpu占用能达到300%多。 所以性能一点不差，至于原理，大学学过的《计算机组成原理》中的cpu流水线技术里面就有讲到，对于单条数据的处理，处理时间就是流过整条流水线的完整过程。对于大量的数据流来说，其平均处理时间，就应该是众多处理环节中最耗时的节点对数据的平时处理时间，而非数据在整条流水线的流过时间。

 

让人着迷的clojure语言

当我试图去深入了解storm时，Storm也为我打开了另一扇门。storm的核心代码是用函数式编程语言clojure写的，由此我便开始接触与学习clojure，尽管没有机会在现在的工作上得以应用，但这门语言已让我着迷了。
将storm和 clojure的用法与思想对照的话，会发现两者在很多方面是不谋而合的。可以说，storm在分式布应用与clojure在语言这两种应用级别上的思想是一致的。也难免作者用使用这么偏门（目前使用者还是较少）的语言进行storm的开发，而且我相信作者用clojure作开发的过程一定相当的爽。
上文中提到storm提供了一种简易的编程模式让我们能够免于线程安全之忧。Clojurer也作了同样的努力，例如将STM（软件事务内存）、Atom 无锁化编程、鼓励不可变值的使用这些思想直接嵌入到语言的语义级别，尽最大努力在语言级别就为你解决这些恼人的问题。
上文还讲到，storm的数据是处理是源源不断，不落地的。如果使用面向对象或者向过程的思想去实现这样的效果，会显得十人痛苦。很可能你为了实现相同的功能，还是会将数据先存储下来，一个步骤做完，再做下一个步骤，这就又回到批处理的流程。然而，如果你用clojure去编写，这一切就会显得自然而又流畅。函数式编程语言，它们让开发者是从更接近于机器的角度（至少语言的设计者是这么认为的）去思考与理解世界，而不是从人的角度去绕弯地设计去多余地构建角色，在函数式编程的世界里面，一切都是围绕数据去进行的，你的所有代码都只不过是对数据的转换。函数即值，函数的层层调用，就是对值的一次次转换；惰性求值，可以让你的map、sequece等结构里面的数据只有在被处理的时候才会计算，数据里面已经内嵌了计算。与面向对象或者面向过程的思考方式不同，你要实现的不再是各种拐弯抹角的逻辑与处理过程，而是最直接的数据转换。而storm让开发者去实现spout与bolt正是数据转换过程。（关于clojure还是其他很多值得细数的特性，今后另开一篇介绍。）

 

其他你需要了解的

本文并没有按部就班的去介绍storm应涵盖的概念与使用方法，只是作一些个人理解层面的闲扯。还有很多部份还未提及，以下稍作列举：

• storm的可靠性保证，ack机制。
• 分布式RPC， storm DRPC。
• Transaction Topology。Storm的事务管理，保证每次处理按序提交，并提交且提交一次。
• TimeCacheMap， RotatingMap。可用于数据做join的场景。
• Trident是Storm之上的高级抽象，类似于hadoop 的pig。

20150405首发于3dobe.com：http://3dobe.com/archives/111/

本站链接接：http://quentinxxz.iteye.com/blog/2199516