转载

Spotify如何对Apache Storm进行规模扩展

【编者的话】Spotify是一家音乐流媒体服务商，最新的数据显示他们已经有6000万用户。 Spotify内部使用 Apache Storm来构建实时类系统，包括广告定位、音乐推荐以及数据可视化等。本文来自Spotify官方技术博客，介绍了Spotify公司如何使用Apache Storm来构建可扩展的个性化系统的。

Spotify已经在很多业务中使用Apache Storm来构建多种实时系统，包括广告定位、音乐推荐以及数据可视化等。其中每一种实时系统都将Apache Storm与其它不同系统加以结合，例如Kafka、Cassandra、Zookeeper以及其它数据输入和输出的系统。Spotify在全球范围内拥有超过五千万的活跃用户，所以在应用构建时需要考虑可扩展性以保证应用的性能以及高可用性。

Thinking Scalability

面对资源线性增长所带来的负载压力提升，可扩展性已经成为软件保持理想性能表现的前提。不过要真正实现可扩展目标，单靠添加资源并对性能进行调整还远远不够。具体而言，可扩展性要求大家对软件方案的设计、质量、可维护性以及性能等方面整体考量。当我们构建应用程序时，首先应该从以下几个方面对可扩展性进行规划（保证应用可扩展性的必要条件）：

软件具备良好的架构与较高的质量。
软件应该易于发布、监控与修改。
软件性能可以跟得上资源线性提升所带来的额外负载增长。

Storm中的可扩展性

那么对Storm流程加以扩展需要在哪些方面做出努力？下面我将通过自己的实时个性化Storm系统作为实例，向大家阐述可扩展能力中的方方面面。

Spotify如何对Apache Storm进行规模扩展

在我们的个性化系统中，使用Kafka集群的Topic（译者注：一个Topic可以认为是一类消息，每个Topic将被分成多个partition）来处理不同类型的事件，比如歌曲完成与广告曝光。我们的个性化Storm拓扑可以订阅不同的用户事件，并将这些事件与读取自Cassandra的实体元数据（例如歌曲流派）相结合，然后将每位用户的事件进行分组，进而通过某种包含聚合与推导机制的算法计算出用户属性。这些用户属性会被写入Canssandra，最后会被多种后端服务使用以提供个性化的用户体验。

设计与质量

当我们将随着时间推移将更多新功能添加到以上个性化流程当中时，我们的拓扑结构开始变得复杂，并直接导致性能调整与事件流调试的难度越来越高。不过较高的测试覆盖率让我们对自身的代码质量充满信心，因此我们认为自己有能力对拓扑进行快速重构并使其投入正常运作。

拓扑架构

在将复杂拓扑转化为小型可维护拓扑的整个转换周期当中，我们通过实际操作得到了以下启示：

为不同任务流创建 小型逻辑拓扑
通过 共享库 而非共享逻辑的方式提高代码的可重复使用度
保证方法容易被测试
并行与批量处理会降低IO操作

质量

我们已经使用Java开发了自己的流程，并借助JUnit对不同计算Bolt内的业务逻辑进行了测试。我们还利用backtype.storm.testing并通过集群模拟进行了端到端的测试。

可维护性

为了将软件轻松部署在集群内的新主机中并对其运行状况进行监控，我们采取了一系列措施来简化维护。

配置

对外暴露所有可调参数，这可以让我们在不变更任何代码的前提下实现软件调整，同时也让我们能够更轻松地实现小型增量变更并观察其实际影响。我们将bolt parallelism、source endpoints、sink endpoints以及其它拓扑性能参数映射到了一个配置文件中。

指标可视化

我们为拓扑指标创建了一套仪表板（dashboard），旨在整体评估其运作状态并进行问题排查。我们采用高级度量指标（详见下图）对整套系统的运行状态加以汇总，因为在面对一套充满了各类指标，但又缺乏重点倾向性的仪表板时，大家往往很难从中找到真正值得关注的信息。

Spotify如何对Apache Storm进行规模扩展

拓扑部署

Spotify如何对Apache Storm进行规模扩展

我们这套个性化流程中的全部计算任务都是幂等的，我们还设计出了自己的部署方案，在事件处理中允许少量的重复以确保部署过程中不会丢失消息。这套方案并不适用于全部用例，特别是在计算任务以事务形式存在的情况下。在以上图表中，各事件由左至右依次排列，其中t1到t8代表着不同时间戳。

在我们的部署方案当中，我们需要确保Storm集群可以同时运行两套个性化拓扑。在t1时间点上，集群运行的是个性化拓扑的v1版本。当我们准备好发布该个性化拓扑的v2版本时，我们会构建并将v2提交至该集群。在t4时间点上，我们的集群正在同时运行这两个版本。每套拓扑都会使用一个唯一的Kafka consumer groupId，从而确保topic内的全部信息都被交付给这两套版本。在此阶段中信息会经过两次处理，但由于计算的幂等属性，这并不会造成任何问题。在t5时间点上，我们停用v1版本，这意味着该版本将不再消费来自Kafka集群的事件。接下来，我们对v2版本的运行图表进行监控，并确保所有指标都处于正常范围之内。如果一切顺利，我们会移除v1版本，并在t8时间点上让集群仅运行v2版本。不过在t7时间上，如果指标图表显示异常状况，我们会激活v1版本并使其延续停用时的状态，继续消费来自Kafka的事件。此时我们还将停用v2版本，这从本质上讲正是我们打造的回滚机制。拥有安全的回滚机制能帮助我们在不断推出小型高频度变更的同时，将相关风险控制在最低程度。

监控与警报

我们会对集群、拓扑、Source、Sink指标加以监控，并为其中的部分高级指标设定警报机制。这是为了避免冗余报警令管理员身心俱疲，甚至忽略掉真正重要的关键性警报。

性能

随着时间的推移，我们已经监控到不同的系统瓶颈与相关的问题，也通过一系列调整将性能维持在理想水平。要获得与预期相符的性能表现，大家还需要选择正确的硬件方案。

硬件

我们最初将自己的拓扑运行在一套共享式Storm集群当中，但随着时间推移，我们发现繁忙的拓扑导致其资源匮乏，并由此引发资源瓶颈。有鉴于此，我们开始使用一套独立的Storm集群，其实这并不算什么难事。现在我们的集群每天要处理超过30亿个事件。整套集群中包含6台主机，每台主机配备24个计算核心、双线程以及32GB内存。即使是使用这套小型集群，我们仍然获得了良好的运行状态。而且在部署过程中需要并行运行2个个性化拓扑版本时，其处理强度与最大利用率仍然相去甚远。未来我们还会考虑在自己的Hadoop集群上将 Storm与YARN 加以结合，从而带来更理想的资源利用率与弹性扩展能力。

吞吐能力与延迟水平

为了获得理想的吞吐能力与延迟水平，我们需要对source与sink参数进行调节。此外我们还做出了其它一系列调整，包括缓存、并行性以及并发性等等，详见下文。

Source和Sink调整

Kafka调整

我们配置了rebalancing.max.tries文件以尽可能减少常见的 rebalance错误。
在每个版本中为不同Kafka Spout采用不同group id，从而确保新拓扑版本在部署过程中遵循冗余信息处理原则。

Cassandra调整

为不同TTL采用不同表。由于无需使用，我们还设置gc_grace_period=0以有效对包含TTL的行组进行读取修复禁用。
使用 DateTieredCompactionStrategy 处理临时性数据。
从Storm拓扑到Cassandra，对开放连接数量加以控制。
配置Snitch以确保调用路由的正确性。

并发问题

Storm中的OutputCollector并非线程安全，也无法保证面向多线程的安全访问——例如对异步处理流程中的Future进行回调。我们利用java.util.concurrent.ConcurrentLinkedQueue对ack/emit Storm tuples的调用进行安全保存，并在bolt内方法执行之初对其加以刷新。

并行调节

我们从 Strata 2014 大会的Storm主题演讲中得到了灵感，进而对拓扑当中的并行机制作出调节。以下几项指导性意见在我们的实例中带来相当出色的表现：

每拓扑每节点1 worker
对于CPU绑定类任务，每计算核心1 executor
对于IO绑定类任务，每计算核心1至10 executor
计算总体并行可能性并将其划分为低速任务与调整任务。低速任务并行性较高，调整任务并行性较低。

Bolt缓存处理

为了维持Bolt当中用户属性计算拥有良好的状态，我们需要从外部与内存两类缓存机制当中做出选择。我们更倾向于使用内存缓存方案，这是因为外部缓存往往会带来网络IO负担，产生不必要的延迟并增加新的故障点。不过说到内存内缓存，我们并没有持久化或者限制内存资源。事实上，我们不太在意持久化，因此只需要着手处理内存限制问题即可。我们最终选择了Guava的Expirable Cache，我们可以在其中定义元素及过期数量上限，从而控制缓存的整体规模。总而言之，这套方案帮助我们在自有集群当中实现了内存的优化利用。

我们通过一整套方案来扩展Storm系统，有了这个系统的支持，我们可以为那些不断增加的活跃用户提供更多的新功能，同时，Storm系统也可以毫无压力地继续为我们提供高可用的服务。