【转】基于Trident构建大规模实时流数据处理系统

刊载于程序员杂志2013年2月号

大数据处理：批量数据处理和流数据处理

Hadoop家族在大数据处理方面的能力已经得到业界认可，但它更适用于对集群上的大数据进行批量处理，并不使用于实时处理大规模流数据。

包括雅虎在内的互联网公司都在尝试建立适用于大规模流数据处理的系统，Twitter的Storm就是其中之一。本文将主要介绍Storm Trident以及雅虎如何基于Storm Trident构建一个流数据处理系统。

Storm基本概念
先介绍本文会涉及到的几个Storm术语。

Storm应用通过Spout从外部数据源读取数据，并将数据以流(stream)的方式发送给Storm节点。用户逻辑运行在Bolt进程。Storm使用Topology(一种有向无环图)来描述所有Spout和Bolt的关系。关于Storm概念细节，请自行查阅相关资料。

Trident特性

Storm是Twitter开发的分布式实时数据处理系统，而Trident是一个基于Storm的用于实时计算的抽象层，详情请参阅这篇文章。

我们可以认为，Trident可以提供如下特性。

• Trident是Storm平台上的“高级语言”(类似于Pig在Hadoop上提供的类SQL查询功能)。Trident支持联结(Join)，聚集(Aggregation)，分组(Group)，函数(Function)和过滤器(Filter)等操作。
• Trident提供对带持久数据层的有状态增量式的数据处理。例如，Trident提供对有状态数据存储到Memcached集群的实现。
• Trident将数据流分成小的被处理(默认设置每个被处理包括1000个事件)，实现了延时和容错(处理失败的时候会被重发)之间的平衡，同时每个批处理只读取持久数据层一次，因此降低了对数据持久层的读写压力。

基于Storm Trident的实时流数据处理系统

为了评估使用Storm作为雅虎下一代实施基础设施的可能性，我们所在的小组构建了一个基于Storm Trident的实时流数据处理系统。该系统处理用户在雅虎网站上的浏览，点击，搜索等事件，实时获取为该用户提供服务的雅虎服务器所在的数据中心，并将这些信息转换成内部编码发送给广告服务系统数据库；同时，为了将这些信息分享给其他系统，我们还对它们进行持久化存储。

应用场景

雅虎在全球各地拥有7亿多的月活跃用户。为了更好地为用户提供服务，雅虎在全球部署了很多数据中心。

雅虎的精准广告投放系统后台流水线根据用户的历史行为计算出用户的属性，然后将用户属性发送给广告服务器数据库。广告服务系统根据数据库中的用户分类属性精准地投放广告。

理想情况下，为了保证广告服务系统能以最快的速度获得用户属性从而更快更准确地相应用户请求，广告服务系统会将用户属性数据库复制到全球各个数据中心。实际上，大部分用户在很长时间内地理位置不会发生变化。例如，台湾地区用户很多时候都会从台湾地区的服务器获取内容，为台湾地区提供广告服务的服务器只需要保存台湾那地区的用户属性就能很好的提供服务了。因此，把美国用户的属性复制到台湾地区的数据中心是没有意义的，而且把用户的属性复制到所有的数据中心是对带宽和存储的巨大浪费。

为了节省服务器的存储和跨数据中心的数据交互，我们会根据用户所在的区域来决定把用户的属性复制到离用户最近的数据中心。例如，台湾地区用户的用户属性只会被复制到与其对应的数据中心的数据库，而不会被复制到美国数据中心。而一旦台湾用户到美国旅游时，我们需要尽快将用户属性复制到美国对应的数据中心。

为了解决上述问题，我们需要实时地从用户事件中抽取对应特征(数据中心的名称)，并将特征转换成用户属性(一种内部编码)，然后将这些编码发送给用户属性数据库，这样数据复制系统会根据这些编码将用户属性复制到对应的数据中心。

系统架构

实时流数据处理系统，包括一个分布式消息队列，用来接收数据采集服务器发送过来的用户事件；一个Storm Topology，用来分布处理用户事件，抽取特征(数据中心的名称)，计算用户属性(数据中心对应的内部编码)，并将最后结果写入持久层和广告服务系统；一个数据持久层服务，用于持久化存储用户属性。该系统的架构如图1所示：

 

数据源和消息队列

Storm允许用户定制Spout从外部数据源读取数据。同时，Storm提供了一些分布式消息队列(例如Kestrel)Spout。

本系统使用Kestrel作为和外部数据源交互的接口。事件收集服务器将用户事件写入Kestrel集群，供Trident Topology读取。

雅虎拥有一套高效，大吞吐量，低延时的数据传输系统—数据高速公路。该系统将用户在Web服务器上的活动日志以事件的方式传递给后台。后台的数据中心接收到事件后，将事件写入Kestrel队列，然后Storm从Kestrel队列读取事件并完成数据处理。

Trident Topology

• 一个Spout从对应的Kestrel集群读取数据，并通过Shuffle Grouping的方式发送给特征抽取Bolt；
• 一个特征抽取Bolt，以无状态方式抽取用户的特征；
• 一个用户计算用户属性的Bolt，根据用户特征计算用户属性，将用户属性持久化存储到数据库，并将用户属性发送给下游系统(广告服务系统)。

在不使用Trident的情况下，我们需要显式地去构建特征抽取Bolt和分类Bolt。通过Trident，我们可以用图2所示的伪代码创建Trident Topology，然后Trident计算出最优的Topology(类似于Pig的查询优化功能)。

为了确保将用户最新的属性发送给广告服务系统，我们必须保证从最新的用户事件中提取信息。Storm为了保证没有数据丢失，一些处理失败的消息会被重复发送，如果我们从这些数据抽取用户属性，就会覆盖掉最新的用户属性。如前文所述，Trident处理数据时会把数据分成小的批处理，并给每个小的批处理分配一个事务ID用户记录该批处理的状态。批处理的ID是根据从数据源读取的顺序递增的，因此当一个失败事务被重发时，对应的批处理ID可能会比最新的批处理小，我们就有机会对这种事务进行特殊的处理。

数据持久层和缓存

为了将提取到的数据信息分享给其他服务，我们必须将用户属性持久化。考虑到雅虎海量的用户属性，我们选择HBase作为数据持久层。

为了保证读的性能，我们为分类器Bolt设计了进程内LRU缓存。由于用户特征在从特征抽取Bolt发送到分类器Bolt采用field grouping的方式，因此同样的用户ID总会被发送到同样的进程，除非Storm Topology被重新平衡(rebalance)，不然Cache不会失效。

数据输出

用户属性除了被持久化存储到HBase之外，同时还会被输出到广告服务系统的数据库。为了避免平滑对广告服务系统数据库的并发写次数，我们将用户属性输出到一个Kestrel队列，然后由广告服务系统从该队列读取用户属性并写入数据库。

结论

基于以上原型系统以及其他团队使用Storm的经验，我们认为，Storm是一个很好的用户构建近实时(Near Realtime)处理应用的系统，具体包括以下几个方面：

• 易于使用，对开发人员也非常友好。开发人员只需要下载Storm包就可以进行开发和调试，而不需要依赖Storm集群。
• 易于部署，维护。
• 活跃的开发社区支持。用户在Storm User Group的问题可以得到及时的回答。
• 丰富的外围支持，例如对分布式消息队列Kestrel的支持，对Memcached的支持。

同时，我们也意识到Storm存在的一些不足。

• 文档不够完整。特别是关于性能优化方面的文档非常少，用户遇到性能问题时，需要尝试很多参数才能找到性能瓶颈。
• 缺少系统级监控。Storm UI提供一些基本的统计信息，包括平均延时，过一段时间处理的事件的数量等，但对于生产系统而言，这些还不够。
• 缺少在集群上进行调试的相关工具。大部分情况下只能通过日志去查找原因，但在生产系统中通常只写很少的日志，因此调试起来比较困难。

（本文作者均来自雅虎北京全球研发中心）