携程这两年在大数据方面有了巨大的发展，数据已经融入了携程每个业务的血液之中，利用大数据和机器学习提升业务已经成为大家的common sense；携程的大数据平台作为支持大数据分析和机器学习的关键系统在这两年之中也有了飞速的增长和进化，在飞速增长的背后如何保证系统的稳定性，解决遇到的问题，并结合业务的需求进行产品的创新对整个大数据平台来说也是很大的挑战。在这个议题中，我将会和大家分享一下这两年来的实践、经验和教训，以及对未来大数据和AI系统的一些想法和展望。

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

一、为什么要做实时数据平台

首先先介绍一下背景，为什么我们要做这个数据平台？其实了解携程的业务的话，就会知道携程的业务部门是非常多的，除了酒店和机票两大业务之外，有近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，HBASE，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上线之后，用户需要一系列的实时指标来观察分流的效果，来确定它配置是否正确；另外需要查看对于订单的影响，如果对订单产生了较大的影响，需要能够及时发现和停止。

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

四、那些曾经踩过的坑

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

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

我们使用的STOM版本是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。

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