去哪儿客户端全业务线用户行为数据ETL介绍

明明如月

2015年7月加入去哪儿，一直从事数据相关工作。热爱数据，忠于数据，敬畏数据。

项目背景

作为旨在聪明消费者出行的 OTA（Online Travel Agent），去哪儿需要开发和完善越来越多的场景来满足用户更加便捷、更加智能和更加人性化的出行需求，这些场景中数据作为最基础的核心价值起着至关重要的作用。而由于业务分工不同，导致不同业务线部门在数据存储清洗和提供服务的方式等各方面都存在着很大的差异。平台公共部门在全业务线业务数据粗粒度收集清洗中做了大量繁重的工作，但是针对在产品设计的众多需要综合利用两个或多个业务线数据的需求中，跨部门之间的细粒度数据交互还是需要数据、算法和开发工程师花费极大精力做很多工作。比如作为全站搜索入口的大搜服务及其衍生的很多个性化推荐需求（猜你喜欢、预制词等）最根本和核心的基础就是需要一份结构化、稳定性和适用性都比较满意的全业务线数据，不需要在每一个项目中都要在数据结构兼容、容灾恢复和过多接口上耗费太多精力。在此前提下，我们依赖于公共部门收集的业务线日志，业务线自己负责输出的后端日志用户行为分析主要包括哪些内容，还有各种补充数据，清洗整理了一份客户端用户主要业务行为的数据，完整刻画了用户行为路径。这份数据有着结构化好、稳定性强和延迟低等众多优点，并以 dubbo 接口的方式提供服务（亦可支持原始数据文件共享）。

数据简介

去哪儿客户端全业务线用户行为数据收集清洗了用户在酒店、机票、度假、门票、火车票、汽车、大搜等 7 个主要业务线或部门的搜索、点击、订单填写、收藏和下单等 5 种行为轨迹，每种行为都定义了至少几十个业务线原始输出或根据应用需求计算得出的字段，并可以根据需要进行扩展。在这份数据中，订单行为包含了用户过去 400 天的数据，其它行为都保存了最近 40 天。每天处理有效行为条数在几千万级别，涉及到的几乎所有行为都会实时的反映到这份数据接口中，从行为产生到接口输出延迟大概在 100 毫秒以内。

流程框架1. 系统概览

如下图所示用户行为分析主要包括哪些内容，用户行为数据 etl 过程主要由离线收集和实时收集两大模块组成。其中离线收集部分每天定时清洗 T-1 天的数据，T-1 到当前的数据收集由实时部分完成。

项目之初我们从离线模块做起，在 hadoop 平台上把每天批量清洗完成的各业务线数据跟之前一天的全量用户行为数据整合，组成截止到当天的全量用户行为数据，结果存放到 hdfs 上供算法和开发人员建模分析使用，这份简单的输出已经给我们一些工作带来很大效益，比如已经应用到猜你喜欢的初始版本。但是随着项目的深入对这份数据提出了更高的要求：

在这些需求的推动下，我们相继开发了实时数据的收集模块和 dubbo 服务接口。实时模块采用 qflume+logstash+kafka+spark streaming 的方式把日志进行收集、清洗和计算并将最终结果存放在 redis 中（qflume 为去哪儿定制化的 flume 版本）。而 dubbo 服务接口是将导入 mysql 的离线模块数据和 redis 中的实时模块数据进行了合并，做出了一份具有实时性的用户数据服务。

实时模块的开发和为除了为项目带来了巨大的收益，比如猜你喜欢模块 ctr 在不做调整算法的情况下就已达到了 100% 左右的提升。此外实时模块让我们对日志的异常状况更加敏感，更能及时的发现和解决问题，而且实时跟离线互为备份，极大增强了用户行为数据的稳定性。

2. 离线框架

离线收集的基础数据以公共收集的 hotdog 和 kylin 日志（hotdog 和 kylin 都为公司的日志服务系统，一些业务线的全部或者部分日志采用了这两个日志服务）以及业务线自己输出的后台日志为主，并辅以其他详情或字典数据，比如常住地信息、酒店详情、景点详情、机场三字码字典等。

具体的处理流程为：业务线各输入数据源首先进行诸如用户黑名单等过滤，并将我们关心的行为日志抽取出来，然后按照一定规则把同一类行为的不同源的日志关联到一起，再从中获取具体相关的行为所需信息，最后需要对行为详情的一些数据进行补充和标准化操作，经过这些步骤后就生成了用户业务线的行为数据。同一天各业务线的行为数据合并后形成当天的用户行为数据（当天，T-1），跟之前历史行为数据（全量，T-2 total），合并成当前最新的用户历史行为数（全量，T-1 total），实际操作中为了减少导入数据库的条数，还生成一份当天的增量用户历史行为数据（T-1 increment），这样每天只把有变化的增量数据更新到数据库里。

每个业务线所依赖数据源不同、生成数据源时间也不同，相应每个业务线的清洗任务都是独立的，每个任务的调度时间和运行时间也都存在着差异，当所有业务线的独立任务调度完成后再统一进行历史和当天用户行为的合并。

离线收集过程特别需要注意的一点是数据源完备性的监测，有时候可能会出现①某台机器几个小时的日志上传延迟或丢失的情况发生（比如上传任务依赖的条件没有满足），或者②服务新上机器后日志并未及时上传。如果这个时候我们清洗任务开始运行并完成最终的合并操作就会导致行为数据缺失，有时候这种缺失并不明显（例如缺失了某台机器一个小时的文件）很难及时发现，在没有上实时任务之前这种异常可能需要很久后的 case 分析才能被追溯回来，挽救措施通常就是花费人力和计算资源进行数据的重新清洗和合并。针对第①个问题，我们增加了输入数据源的机器列表和小时文件数和大小的检测，不满足条件则延迟调度，失败次数超过阈值人工去干预；而第②个问题，直到我们上了实时模块后，通过行为条数的统计对比才彻底解决。

下面以度假业务线为例，介绍一下从业务线日志到最终合并到用户行为数据处理的整个流程：第一步对 hotdog 日志、accesslog 日志和 PostgreSQL 数据库里的多个数据源输入进行完备性检测，每个数据源包含一到多种行为的数据详情，比如 hotdog 中会包含搜索、详情页、收藏和下订单的数据；第二步对这些不同源的数据需要进行黑名单、有效标识等过滤规则过滤，并抽取出我们关心的行为数据进行分类，这个过程中我们不关心的行为的日志也会被丢弃；第三步对于每一类行为日志，并按照一定规则进行碰撞，从而将不同数据源中同一类行为的详情合并到一起（hotdog 中的订单和 PostgreSQL 中的订单），接下来就可以从合并的日志行为获取中所需要的字段，并进行不同标准下的 id 数据加解密和详情的关联，以及诸如用户常住地、城市化标准化名等信息的补全和标准化；第四步生成度假业务线的用户行为数据；最后一步就是全业务线数据的合并和入库。整个过程会嵌入我们对数据指标的统计存储和关键信息的监控，统计结果会存入 mysql 并在内部统计系统上展示。

3. 实时框架

实时计算由于其高效低延迟等特性已经在业界有着广泛的应用，在整个数据处理和分析领域有着举足轻重的地位。通常来说实时计算包含采集与传输、分析与计算和存储与服务三个阶段，每个阶段都有若干成熟的开源框架。在我们的系统中，数据采集使用的是去哪儿定制化的 flume 版本——qflume，部署在业务线生产机器上 qflume agent 将采集到的日志输出到 kafka 中，根据业务应用的不同收集到不同 topic，其中有些 topic 会经过 logstash 进行预处理，做一些粗粒度的无效信息过滤和一些规范化的输出，logstash 清洗后的数据依然是输出到 kafka 中去。kafka 中具体某一 topic 的数据会作为输入在实时计算平台上进行更复杂逻辑的处理，这里的实时计算平台我们采用的 spark streaming 框架。spark streaming 在进行复杂逻辑计算的时候，有些中间结果数据需要缓存，缓存数据量在几十 G 的规模，请求 QPS 约几十 K，加上我们最终结果也是大致相同量级的输出和读写请求，核算下来整个系统需要的缓存空间约百 G，峰值 QPS 达到 100K+，经过调研尝试我们最终选择了 redis 作为缓存服务。

具体以酒店业务线为例，hotdog 和 kylin 原始 topic 经过 logstash 过滤出所需要处理的酒店搜索点击等行为相关的日志，并对日志中部分无效信息进行清理，最后得出相对轻量级和规范的 topic，其中由于业务架构因素 hotdog 日志源比较复杂，需要进行多次清洗合并。清洗后的 hotdog 和 kylin 两种类型数据的 topic 经过 spark streaming 分发碰撞（碰撞在 redis 中缓存），不同数据源中相同的 traceId 会 join 到一起，从而组成同一会话行为的全部数据，然后进行复杂逻辑的处理，清洗出所要收集行为的字段详情，部分字段还需要通过其他数据源进行补充和规范，比如日志中一些下单酒店的详细信息，需要我们每天从异步加载更新的酒店详情里面获取补充，每条行为的城市也都要经过标准化处理。

4. 接口及服务

我们提供了 dubbo 接口的线上服务，同时也支持原始 hdfs 文件的共享。原始结果数据以 json 格式组织，并封装了 jar 去解析，为了提高响应速度线上服务接口采用了 Protobuffer 的序列化和反序列化方式，并封装在 jar 包里对外透明，用户可以只关心调用的方法。

5. 异常及优化

为了保证我们数据的稳定性和有效性，我们针对实际中出现的很多问题进行了监控和优化，这些举措除了提升了我们数据的可用性之外，很多时候也能让我们先于业务线或计算平台发现问题并反馈给他们。具体说来，主要有以下几个方面：

1) 实时处理过程中要对 kafka 的 delay、日志 receive count、清洗结果 count，某些异常 exception count 等关键数据进行监控，设置阈值，实时报警，这个监控我们在公司 watcher 平台上完成。

2) 离线数据源要进行完备性检测，并且对每天的离线结果数据和实时结果数据进行统计对比，差异的合理范围在万分位到千分位之间，1% 以上的差异就会引起警觉并去排查，这一步在我们内部统计系统 Datamind 有报表和报警。

3) 我们提供的数据按用户 id 进行聚集，服务接口也是按 id 进行索引的，但有些业务场景需要知道过去几秒内发生某种行为的一批 id，比如要对刚刚下了机票订单的用户进行接送机服务消息推送，就需要知道过去几分钟下机票订单的用户，针对这个场景，我们在实时处理一条用户的行为数据时，同时也对发生这个行为的 id 进行了实时记录，并有服务进行实时获取和推送。

4) 实时数据 redis 设置了 48 小时失效期，但并不是 48 小时前的实时数据都扔掉，对于过去 10 天的 redis 数据我们都进行持久化到了本地，一方面能对离线 T-1 甚至更久的数据异常时进行备份，另一方面对于离线实时差异分析具有极其重要的作用，能还原实时现场。

5) 爬虫或者打接口的 id 都是在没有预警的情况下发生，这两种情况引起的大量的请求容易导致数据量超大把实时任务卡死的情况，我们在实时处理的时候进行了异常监测和黑名单处理，同时为了保证实时和离线数据的一致性，离线处理时每天同步最新的黑名单数据进行过滤。

6) 对于数据数据源对应多个服务产生或者数据源日志异常（比如单条超过 10M）的情况我们通常利用 logstash 在 kafka 的 topic 级别进行清洗合并。

7) 每天全量结果文件百 G 级别，直接入库时间长压力大，对此我们进行了差量化计算，得出每天的差量再入库，极大的降低了数据库读写的请求。

应用简介

这份数据上线之后用于推荐和建模分析非常有效，已经应用到集团内 30 多个线上线下项目当中，其中典型的应用包括：

经验总结

从这份数据立项到稳定使用，在整个数据收集处理以及应对业务线日志格式变更、日志收集，任务优化等方面做了大量工作，也增加了很多切身体会：