携程基于 Storm 的实时大数据平台实践

本文讲解了携程在实时数据平台的一些实践，按照时间顺序来说明我们是怎么一步一步构建起这个实时数据平台的，目前有一些什么新的尝试，未来的方向是怎么样的，希望对需要构建实时数据平台的公司和同学有所借鉴。

为什么要做实时数据平台

首先先介绍一下背景，为什么我们要做这个数据平台？其实了解携程的业务的话，就会知道携程的业务部门是非常多的，除了酒店和机票两大业务之外，有近 20 个 SBU 和公共部门，他们的业务形态差异较大，变化也快，原来那种 Batch 形式的数据处理方式已经很难满足各个业务数据获取和分析的需要，他们需要更为实时地分析和处理数据。

其实在这个统一的实时平台之前，各个部门自己也做一些实时数据分析的应用，但是其中存在很多的问题：

首先是技术选型五花八门，消息队列有用 ActiveMQ 的，有用 RabbitMQ 的，也有用 Kafka 的，分析平台有用 Storm 的，有用 Spark-streaming 的，也有自己写程序处理的；由于业务部门技术力量参差不齐，并且他们的主要精力还是放在业务需求的实现上，所以这些实时数据应用的稳定性往往难以保证。

其次就是缺少周边设施，比如说像报警、监控这些东西。

最后就是数据和信息的共享不顺畅，如果度假要使用酒店的实时数据，两者分析处理的系统不同就会很难弄。所以在这样前提下，就需要打造一个统一的实时数据平台。

需要怎样的实时数据平台

这个统一的数据平台需要满足4 个需求：

首先是稳定性，稳定性是任何平台和系统的生命线；

其次是完整的配套设施，包括测试环境，上线、监控和报警；

再次是方便信息共享，信息共享有两个层面的含义，1、是数据的共享；2、是应用场景也可以共享，比如说一个部门会受到另一个部门的一个实时分析场景的启发，在自己的业务领域内也可以做一些类似的应用；

最后服务响应的及时性，用户在开发、测试、上线及维护整个过程都会遇到各种各样的问题，都需要得到及时的帮助和支持。

如何实现

在明确了这些需求之后我们就开始构建这个平台，当然第一步面临的肯定是一个技术选型的问题。消息队列这边 Kafka 已经成为了一个既定的事实标准；但是在实时处理平台的选择上还是有蛮多候选的系统，如 Linkedin 的 Samza， apache 的 S4，最主流的当然是 Storm 和 Spark-streaming 啦。

出于稳定和成熟度的考量，当时我们最后是选择了Storm 作为实时平台。如果现在让我重新再来看的话，我觉得 Spark-streaming 和 Storm 都是可以的，因为这两个平台现在都已经比较成熟了。

架构图的话就比较简单，就是从一些业务的服务器上去收集这个日志，或者是一些业务数据，然后实时地写入 Kafka 里面，Storm 作业从 Kafka 读取数据，进行计算，把计算结果吐到各个业务线依赖的外部存储中。

那我们仅仅构建这些就够了吗？当然是远远不够的，因为这样仅仅是一些运维的东西，你只是把一个系统的各个模块搭建起来。

前面提到的平台的两个最关键的需求：数据共享和平台整体的稳定性很难得到保证，我们需要做系统治理来满足这两个平台的关键需求。

首先 **** 说说数据共享的问题，我们通常认为就是数据共享的前提是指用户要清晰的知道使用数据源的那个业务含义和其中数据的 Schema，用户在一个集中的地方能够非常简单地看到这些信息；我们解决的方式是使用 Avro 的方式定义数据的 Schema，并将这些信息放在一个统一的 Portal 站点上；数据的生产者创建 Topic，然后上传 Avro 格式的 Schema，系统会根据 Avro 的 Schema 生成 Java 类，并生成相应的 JAR，把 JAR 加入 Maven 仓库；对于数据的使用者来说，他只需要在项目中直接加入依赖即可。

此外，我们封装了 Storm 的 API，帮用户实现了反序列化的过程，示例代码如下，用户只要继承一个类，然后制定消息对应的类，系统能够自动完成消息的反序列化，你在 process 方法中拿到的就是已经反序列化好的对象，对用户非常方便。

其次 **** 我们来说说资源控制，这个是保证平台稳定性的基础，我们知道 Storm 其实在资源隔离方面做得并不是太好，所以我们需要对用户的 Storm 作业的并发做一些控制。我们的做法还是封装 Storm 的接口，将原来设定 topology 和 executor 并发的方法去掉，而把这些设置挪到 Portal 中。下面是示例的代码：

另外，我们前面已经提到过了，我们做了一个统一的 Portal 方便用户管理，用户可以查看 Topic 相关信息，也可以用来管理自己的 Storm 作业，配置，启动，Rebalance，监控等一系列功能都能够在上面完成。

在完成了这些功能之后，我们就开始初期业务的接入了，初期业务我们只接了两个数据源，这两个数据源的流量都比较大，就是一个是 UBT（携程的用户行为数据），另一个是 Pprobe 的数据（应用流量日志），那基本上是携程用行为的访问日志。主要应用集中在实时的数据分析和数据报表上。

在平台搭建的初期阶段，我们有一些经验和大家分享一下：

1. 最重要的设计和规划都需要提前做好，因为如果越晚调整的话其实付出的成本会越大的；
2. 集中力量实现了核心功能；
3. 尽早的接入业务，在核心功能完成并且稳定下来的前提下，越早接入业务越好，一个系统只有真正被使用起来，才能不断进化；
4. 接入的业务一定要有一定的量，因为我们最开始接入就是整个携程的整个 UBT，就是用户行为的这个数据，这样才能比较快的帮助整个平台稳定下来。因为你平台刚刚建设起来肯定是有各种各样的问题的，就是通过大流量的验证之后，一个是帮平台稳定下来，修复各种各样的 bug，第二个是说会帮我们积累技术上和运维上的经验。

在这个之后我们就做了一系列工作来完善这个平台的“外围设施”：

首先就是把 Storm 的日志导入到 ES 里面，通过 Kanban 展示出来；原生的 Storm 日志查看起来不方便，也没有搜索的功能，数据导入 ES 后可以通过图标的形式展现出来，也有全文搜索的功能，排错时非常方便。

其次就是 metrics 相关的一些完善；除了 Storm 本身 Build in 的 metrics 之外我们还增加了一些通用的埋点，如从消息到达 Kafka 到它开始被消费所花的时间等；另外我们还是实现了自定义的 MetricsConsumer，它会把所有的 metrics 信息实时地写到携程自己研发的看板系统 Dashboard 和 Graphite 中，在 Graphite 中的信息会被用作告警。

第三就是我们建立了完善的告警系统，告警基于输出到 Graphite 的 metrics 数据，用户可以配置自己的告警规则并设置告警的优先级，对于高优先级的告警，系统会使用 TTS 的功能自动拨打联系人的电话，低优先级的告警则是发送邮件；默认情况下，我们会帮用户添加 Failed 数量和消费堵塞的默认的告警。

第四，我们提供了适配携程 Message Queue 的通用的 Spout 和写入 Redis，HBbase，DB 的通用的 Bolt，简化用户的开发工作。

最后我们在依赖管理上也想了一些方法，方便 API 的升级；在 muise-core（我们封装的 Storm API 项目）的 2.0 版本，我们重新整理了相关的 API 接口，之后的版本尽量保证接口向下兼容，然后推动所有业务都升级一遍，之后我们把 muise-core 的 jar 包作为标准的 Jar 包之一放到每台 supervisor 的 storm 安装目录的 lib 文件夹下，在之后的升级中，如果是强制升级，就联系用户，逐个重启 Topology，如果这次升级不需要强制推广，等到用户下次重启 Topology 时，这个升级就会生效。

在做完这些工作之后，我们就开始大规模的业务接入了，其实目前基本上覆盖了携程的所有的技术团队，应用的类型也比初期要丰富很多。

下面给大家简单介绍一下，在携程的一些实时应用；

主要分为下面四类：

1. 实时数据报表；
2. 实时的业务监控；
3. 基于用户实时行为的营销；
4. 风控和安全的应用。

第一个展示的是携程这边的网站数据监控平台 cDataPortal，携程会对每个网页访问的性能做一些很详细的监控，然后会通过各种图表展示出来。

第二个应用是携程在 AB Testing 的应用，其实大家知道 AB Testing 只有在经过比较长的一段时间，才能得到结果，需要达到一定的量之后才会在统计上有显著性；那它哪里需要实时计算呢？实时计算主要在这边起到一个监控和告警的作用：当 AB Testing 上线之后，用户需要一系列的实时指标来观察分流的效果，来确定它配置是否正确；另外需要查看对于订单的影响，如果对订单产生了较大的影响，需要能够及时发现和停止。

第三个应用是和个性化推荐相关，推荐其实更多的是结合用户的历史偏好和实时偏好来给大家推荐一些场景。这边实时偏好的收集其实就是通过这个实时平台来做的。比较相似的应用有根据用户实时的访问行为推送一些比较感兴趣的攻略，团队游会根据用户的实时访问，然后给用户推送一些优惠券之类的。

那些曾经踩过的坑

在说完了实时数据平台在携程的应用，让我们简单来聊聊这个过程中我们的一些经验。

首先是技术上的，先讲一下我们遇到的坑吧。

我们使用的 Storm 版本是 0.9.4，我们遇到了两个 Storm 本身的 BUG，当然这两个 bug 是比较偶发性的，大家可以看一下，如果遇到相应的问题的话，可以参考一下：

• storm-763：Nimbus 已经将 worker 分配到其他的节点，但是其他 worker 的 netty 客户端不连接新的 worker；

应急处理：Kill 掉这个 worker 的进程或是重启相关的作业。

• storm-643：当 failed list 不为空时，并且一些 offset 已经超出了 Range 范围，KafkaUtils 会不断重复地去取相关的 message；

另外就是在用户使用过程中的一些问题，比如说如果可能，我们一般会推荐用户使用 localOrShuffleGrouping，在使用它时，上下游的 Bolt 数要匹配，否则会出现下游的大多数 Bolt 没有收到数据的情况，另外就是用户要保证 Bolt 中的成员变量都要是可序列化的，否则在集群上运行时就会报错。

然后就是关于支持和团队的经验，首先在大量接入前其告警和监控设施是必须的，这两个系统是大量接入的前提，否则难以在遇到非常问题时及时发现或是快速定位解决。

第二就是说清晰的说明、指南和 Q&A 能够节约很多支持的时间。用户在开发之前，你只要提供这个文档给他看，然后有问题再来咨询。

第三就是要把握一个接入节奏，因为我们整个平台的开发人员比较少，也就三个到四个同学，虽然已经全员客服了去应对各个 BU 的各种各样的问题，但是如果同时接入太多项目的话还会忙不过来；另外支持还有重要的一点就是“授人以渔”，在支持的时候给他们讲得很细吧，让他们了解 Kafka 和 Storm 的基本知识，这样的话有一些简单问题他们可以内部消化，不用所有的问题都来找你的团队支持。

新的探索

前面讲的是我们基本上去年的工作，今年我们在两个方向上做了一些新的尝试：Streaming CQL 和 JStorm，和大家分享下这两个方面的进展：

Streaming CQL是华为开元的一个实时流处理的 SQL 引擎，它的原理就是把 SQL 直接转化成为 Storm 的 Topology，然后提交到 Storm 集群中。它的语法和标准的 SQL 很接近，只是增加了一些窗口函数来应对实时处理的场景。

下面我通过一个简单的例子给大家展示一个简单的例子，给大家有个直观的感受。我的例子是

• 从 kafka 中读取数据，类型为 ubt_action；
• 取出其中的 page，type，action，category 等字段然后每五秒钟按照 page, type 字段做一次聚合；
• 最后把结果写到 console 中。

如果需要用 Storm 实现的话，一般你需要实现 4 个类和一个 main 方法；使用 Streaming CQL 的话你只需要定义输入的 Stream 和输出的 Stream，使用一句 SQL 就能实现业务逻辑，非常简单和清晰。

那我们在华为开源的基础上也做了一些工作：

• 增加 Redis，Hbase，Hive（小表，加载内存）作为 Data Source；
• 增加 Hbase，MySQL / SQL Server，Redis 作为数据输出的 Sink；
• 修正 MultiInsert 语句解析错误，并反馈到社区；
• 为 where 语句增加了 In 的功能；
• 支持从携程的消息队列 Hermes 中读取数据，

Streaming CQL 最大的优势就是能够使不会写 Java 的 BI 的同事，非常方便地实现一些逻辑简单的实时报表和应用，比如下面说到的一个度假的例子基本上 70 行左右就完成了，原来开发和测试的时间要一周左右，现在一天就可以完整，提高了他们的开发效率。

【案例】度假 BU 需要实时地统计每个用户访问“自由行”、“跟团游”、“半自助游”产品的占比，进一步丰富用户画像的数据：

• 数据流：UBT 的数据；
• Data Source：使用 Hive 中的 product 的维度表；
• 输出：Hbase。

今年我们尝试的第二个方向就是 Jstorm，Storm 的内核使用 Clojure 编写，这给后续深入的研究和维护带来了一定的困难，而 Jstorm 是阿里开源的项目，它完全兼容 storm 的编程模型，内核全部使用 Java 来编写，这就方便了后续的研究和深入地调研；阿里的 Jstorm 团队非常 Open，也非常专业化，我们一起合作解决了一些在使用上遇到的问题；除了内核使用 Java 编写这个优势之外，Jstorm 对比 storm 在性能上也有一定的优势，此外它还提供了资源隔离和类似于 Heron 之类的反压力机制，所以能够更好的处理消息拥塞的这种情况。

我们现在基本上已经把三分之一的storm 应用已经迁到 Jstorm 上了，我们使用的版本是 2.1；在使用过程中有一些经验跟大家分享一下：

第一点是我们在与 kafka 集成中遇到的一些问题，这些在新版本中已经修复了：

• 在 Jstorm 中，Spout 的实现有两种不同的方式：Multi Thread（nextTuple，ack & fail 方法在不同的进程中调用）和 Single Thread，原生的 Storm 的 Kafka Spout 需要使用 Single Thread 的方式运行；
• 修复了 Single Thread 模式的 1 个问题（新版本已经修复）。

第二点是 Jstorm 的 metrics 机制和 storm 的机制完全不兼容，所以相关的代码都需要重写，主要包括适配了 Kafka Spout 和我们 Storm 的 API 中的 Metrics 和使用 MetricsUploader 的功能实现了数据写入 Dashboard 和 Graphite 的功能这两点，此外我们结合了两者的 API 提供了一个统一的接口，能兼容两个环境，方便用户记录自定义的 metrics。

以上就是我要分享的内容，在结尾处，我简单总结一下我们的整体架构：

底层是消息队列和实时处理系统的开源框架，也包括携程的一些监控和运维的工具，第二层就是 API 和服务，而最上面通过 Portal 的形式讲所有的功能提供给用户。

未来方向

在分享的最后，我来和大家聊聊实时数据平台未来的发展方向，主要有两个：

1. 继续推动平台整体向 Jstorm 迁移，当然我们也会调研下刚刚开源的 Twitter 的 Heron，与 Jstorm 做一个对比；
2. 对于 dataflow 模型的调研和落地，去年 google 发表了 dataflow 相关的论文（强烈建议大家读读论文或是相应的介绍文章），它是新一代实时处理的模型，能在保证实时性的同时又能保证数据的正确性，目前开源的实现有两个：Spark 2.0 中 Structured Streaming 和 Apache 的另一个开源项目 BEAM，BEAM 实现了 Google Dataflow 的 API，并且在 Spark 和 Flink 上实现了相应的 Executor。

下半年我们还会做一些调研和探索性的尝试，并寻找合适的落地场景。

作者介绍

张翼，携程大数据平台负责人，浙江大学硕士毕业，2015 年初加入携程，主导了携程实时数据计算平台的建设，以及携程大数据平台整合和平台技术的演进。进入互联网行业近 10 年，从事大数据平台和架构的工作超过 6 年。

感谢杜小芳对本文的审校。

给InfoQ 中文站投稿或者参与内容翻译工作，请邮件至 editors@cn.infoq.com 。也欢迎大家通过新浪微博（ @InfoQ ， @丁晓昀），微信（微信号： InfoQChina ）关注我们。