AI时代生产资料从何而来？打造千万级QPS规模实时大数据收集体系

作者Weichen, 是爱奇艺云平台大数据采集业务负责人。2014年加入爱奇艺云平台，在大数据的生产/采集/传输/分析相关工作上有一定的经验。

---

大数据，是近年来互联网行业内最火热最高频的技术名词，它的应用和影响甚至已经远远超出了互联网行业，渗透到社会和日常生活中的各个领域。海量的数据，尤其是用户数据，其本身已经变成了一种企业和社会关注的重要战略资源，成为大家争相抢夺的焦点。就像农业时代的土地，工业时代的能源，数据就是未来互联网AI时代最重要的基本生产资料。

那么，这些海量数据究竟从何而来？

从内容层面来说，互联网企业的大数据，主要来源于海量的用户行为和程序行为产生的记录。企业通过打造一套强有力的数据收集体系，可以实时高效的，将所有用户行为/服务器程序行为，进行收集汇总，进行清洗和过滤，再提供给后续的数据分析和挖掘使用。

建设实时的大数据收集体系, 对于一些用户规模极大，尤其是用户使用产品时间极长的大型互联网公司而言，是一项非常大的技术挑战，爱奇艺就是一个这样的典型。目前爱奇艺的用户和机器，峰值每秒生产日志超过一千万行，每天生产的日志数据总量超过3千亿行。这个数据规模是个什么概念呢？如果用5号字把日志一行行打印在纸上，纸张的高度每小时可以绕地球一周。

爱奇艺的实时大数据收集体系，正是为了应对这样海量的数据收集场景而生，它提供了千万QPS级的实时采集能力，收集了爱奇艺各端所有用户行为日志和后台服务器程序日志，为爱奇艺的各种数据产品提供了最重要的数据来源支撑。

骐骥千里，非一日之功。爱奇艺的实时大数据收集体系，也经历了长期的发展和迭代。

第一代数据收集体系(2013-2014)

最初的数据收集体系，主要是围绕早期的简单大数据场景，如统计用户数据，分析产品运营报表等需求而设计。

对于用户数据的统计类需求，第一步是在用户终端通过代码埋点的方式，将用户的使用行为经过HTTP请求投递到后台Nginx服务器，让Nginx服务器通过打印日志的方式记录用户使用产品的详细信息。这些Nginx日志被定期汇总，整理并上传到HDFS，供后续大数据分析和报表计算使用。

图1：第一代用户数据采集架构

初代数据收集方案的主要缺点是，数据从生产到使用的延迟较高。用户在客户端上的行为触发了HTTP投递后，大约10~15分钟后才能被上传到HDFS上。在这样的数据收集方案下，只能满足业务方一些实时性要求不高的需求，例如统计昨日的用户数据，绘制运营报表等。

对于一些有实时性需求的大数据场景，比如根据用户的历史行为，对用户进行画像，进行精准的内容和广告推荐。在这个旧的收集机制下，推荐内容都是基于用户昨天以前的行为进行分析得到的，用户实时的操作和行为不能反映到他的产品体验中。从这个角度看，正是数据采集的实时性不足，从源头上限制了推荐类产品能达到的效果。

第二代数据收集体系(2015-2016)

随着公司业务的不断发展，对数据收集体系提出了新的要求，首当其冲的就是需要一个更快速，能实现秒级延迟的实时数据采集方案。另外，在数据来源上，除了收集用户数据之外，也需要增加对机器日志采集分析的支持。来自后台机器的程序日志数据，可以帮助各个技术团队完成技术故障报警，故障诊断排查，性能指标统计等等需求。经过开发迭代，第二代的实时数据采集体系如下所示：

图2：用户数据采集+程序日志采集

在引入了实时数据采集之后，业务方在使用数据的形式上，就有了灵活性的选择。例如各种报表监控统计的场景，可以先用实时数据统计绘制实时报表，用于观察实时的趋势；在隔日再通过HDFS数据离线计算获得精准的结果。同时使用实时数据和离线数据，离线数据的计算结果作为对实时数据统计结果的修正和补充 ( 业界俗称大数据处理的Lambda架构 )。

更重要的是，实时数据源的提供，为一些基于用户大数据分析的产品设计，带来的新的思路。例如内容和广告的推荐，可以使用实时采集的数据进行计算，触发算法推荐，让用户上一时刻的操作行为，很快能反应到下一时刻的产品体验上。又例如安全风控环节，可以根据实时的用户行为日志，即刻捕获到一些在进行安全威胁操作的账号和IP，对它们进行快速的拦截。

在第二代数据收集体系中，我们引入了实时数据的采集器Apache Flume和分布式消息队列Kafka. 社区开源的Apache Flume，支持通过用户自定义的脚本的监听日志文件，将监听到的日志文件，实时写入到Kafka消息队列中。如下是一个基于Flume监听文件的示例。

图3: 基于Flume-Kafka 的实时数据采集架构

引入了开源Flume和Kafka之后，数据收集体系在一定程度上满足了部分业务的实时数据需求。但是，随着数据采集业务规模的不断扩大，新的问题不断暴露出来，主要体现在以下几个方面：

1.不同业务的日志采集，需要定义不同的采集脚本和Flume采集配置，当业务发展到一定规模后，维护管理成本巨大。

2.对实时采集配置的任何修改，都要伴随着大规模线上集群Flume重启等运维操作。

3.Apache Flume自身的功能限制，不能满足灵活多变的业务需求，如按比例随机采样，数据加工，数据格式转换等等。

第三代数据收集体系(2016-2017)

自2016年以来，随着公司业务的不断规模化多样化，产生了更多新的大数据业务需求，也带来了更高的运维和管理负担。因此，数据采集的自动化，规模化，以及可管理性变得尤为重要。同时，日益增长的大数据业务场景，对数据的质量提出了更高的要求。业务组希望使用的数据是经过清洗加工筛选过后的，直接符合业务场景的数据，而非刚采集上来的原始毛坯数据。另外，业务对数据来源的需求也更加广泛，需要支持数据库产品中的binlog日志收集，以及Docker类弹性容器中的日志数据收集。

在这些新需求的推动下，爱奇艺的大数据收集体系也在进一步的迭代：

图4: 新版大数据实时采集体系

新的数据收集体系，最主要的有三个改变：

1.数据采集的过程高度自动化，大幅降低人工运维成本。

2.数据收集的目标形态(离线HDFS和实时Kafka)可以不受数据来源的限制。对数据的使用方而言，数据的来源是黑盒。

3.既能提供原始采集数据，也能提供加工清洗筛选后的数据。

为了实现这些改变，在技术上主要有两个大的突破，下面分别来介绍一下。

1.自主研发的实时数据采集客户端Venus-Agent

2.通过Apache Flink实现了数据的实时加工层

技术突破 1

数据采集客户端Venus-Agent

在设计第三代数据收集体系的过程中，我们首先考虑到的是数据采集工作的自动化。包括Apache Flume在内的大量开源软件，其启动方式都是读取本地配置文件进行启动。我们最初试图通过对Flume源码进行改造，将所有采集客户端的配置集中管理起来，让管理员可以有一个统一入口，管理所有数据源的采集配置。

在沿着这个思路进行Flume源码改造的过程中，我们发现了Apache Flume自身存在的一些设计缺陷和鲁棒性问题。此外，由于开源Flume版本更迭较慢，在一些复杂的业务场景（如Docker类弹性资源池）上，Flume甚至无法满足基本的数据采集需求。我们最终决定，只仿照Apache Flume的功能框架，自行重新开发一套全新的数据实时采集客户端Venus-Agent。

重新设计的数据采集客户端Venus-Agent，相较于传统的Apache Flume采集，有几个大的变动：

变动1：不再使用tail 类命令或脚本监听固定文件，而是直接通过INode号码锁定需要采集的文件。在采集文件的选取上，通过定期Check采集配置和检测文件目录，实时更新采集文件的列表。

基于这个变动，我们实现了在各种极端情况下的数据采集。尤其是在类似Docker弹性容器环境中，当用户应用的容器数量的增减伸缩，容器上的日志采集也能随之同步伸缩。弹性资源池的用户，只需要关心它的应用容器的扩容缩容，无需关心扩容后新增容器里的日志采集问题。下图演示的是Venus-Agent在Docker类弹性容器资源池场景上，支持日志采集随着容器扩容缩容同步变化。

图5：弹性容器资源池里的日志采集

第二个大的变化是:

变动2：实时记录每个文件采集的进度Offset，即使发生网络抖动/后端Kafka故障/采集进程暂停等等异常情况，在异常恢复后，Venus-Agent还能从之前的Offset记录位置继续采集，保持消息不丢失。

通过这个特性的加入，我们大幅提升了数据采集的鲁棒性。让采集客户端可以抵抗一定程度的极端情况风险。在小型风险事故发生时，采集中断，原始采集文件上的Offset偏移量停止移动。当故障异常恢复后，Venus-Agent可以自动沿着之前的Offset记录位置继续向下采集，保证数据没有丢失。

图6：通过Offset提升数据采集的鲁棒性

第三个变化也是最关键的变化：

变动3：所有客户端的数据采集的配置全部由远端的统一平台保存管理。采集客户端通过定期心跳的方式，向后端拉取最新的配置，并实时生效。

图7：客户端从后端系统拉取采集配置

在新版Venus-Agent采集客户端中，我们实现了采集配置的远程集中管理，这给我们在数据采集业务的运维管理上带来了极大的便利。例如，修改数据的采集配置，变更数据的采集目标等等一系列常见操作，都可以由用户自助的在后台系统的Web页面上操作。当页面上修改采集配置之后，经过一个心跳间隔，部署在日志服务器上的采集客户端就会通过心跳拉取到最新的数据采集配置，并立刻生效。新的采集客户端解决了Apache Flume给我们带来的数据管理和运维管理问题，极大的减轻了团队的运维压力，提升了业务推进效率。

如下图所示，在新版Venus-Agent上线后的第二个季度，单个季度新增采集的机器数量，就超过了过去一年部署Apache Flume采集的机器数量的总和。

图8：Venus-Agnet自动化管理提升业务推进效率

技术突破 2

实时数据的筛选加工模块

随着实时业务的不断发展壮大，越来越多的团队提出了数据加工和清洗的需求。以用户行为的实时数据采集为例，在旧的做法中，往往是直接向业务部门提供一个大而全的Kafka数据流，包含全部的用户行为数据。由业务部门自行进行过滤清洗，筛选出自己业务需要的一小部分数据。这种粗放的数据提供方式，让各个业务线在进行实时数据分析的过程中，重复的消耗了大量的计算资源。最初为了解决这个问题，提出了两个临时办法。

临时方案 1：

按日志内容中的特定字段进拆分过滤

在对采集到的数据流进行解析后，按照特定字段的值做区别，分别把数据发向多个不同的Kafka Topic （该方案相当于Apache Flume中的Selector机制）。该方案可以用下图简单描述。

图9：按照指定字段的值，拆分筛选数据流

该方案的主要问题是，面对日益复杂的互联网数据分析需求，经常会有多个团队需求的数据出现维度交叉，数据需要被多个团队重复使用的情况。例如：

1.A业务需要爱奇艺苹果IPhone用户播放行为数据

2.B业务需要爱奇艺用户在北京联通网络环境下的播放行为数据

3.C业务需要爱奇艺动漫频道里用户的播放行为数据

三个团队的数据需求中有交叉的部分，对于原始采集的一份爱奇艺用户行为的原始数据流，要生产加工出上面三种业务需要的指定数据流，显然用直接按字段拆分的方法无法实现。

旧方案 2：

重复采集+分别筛选加工

为了解决上述方案遇到的维度冲突问题，不得不在一些业务的采集环节，重复采集多次原始数据，来应对上面提到的维度冲突问题。在原始的Apache Flume时代，这些重复采集+重复过滤，甚至是用脚本完成的。下图是一个示例：

图10：在Apache Flume上

通过脚本采集并筛选数据示例

在2015~2016年度，我们曾经在生产环境上大量的使用这个方案，很快遇到一个瓶颈，即这些用于过滤筛选数据的grep/awk脚本，消耗的是日志打印所在的线上机器上的CPU资源。当每一台机器上运行了几十个这样的grep/awk脚本之后，CPU达到瓶颈，后续就无法再新增脚本了。这个瓶颈限制了我们实时数据的生产规模的扩大，拖累了各个团队实时计算业务的发展速度。我们下定决心，将数据的加工，清洗，筛选工作从采集端剥离出来，使用专门的数据加工层来进行处理，并确保数据加工层的吞吐能力是可水平扩展的。

目前现行的实时数据采集+加工方案架构如下：

1.设计双层Kafka结构，第一层Kafka接收采集端上传的原始全量数据。

2.使用Flink实时计算集群进行数据加工计算

3.Flink层计算逻辑全部保存在后端管理系统里。Flink实时计算任务中，数据筛选/加工/过滤的逻辑，通过定期心跳向后端管理系统获取。到Flink任务中实时生效。

在这个方案中，各个团队无论需求的数据加工逻辑如何复杂多样，对数据筛选的维度有无冲突，都可以在Flink实时计算层进行处理。同时，由于Flink计算加工的逻辑全部从后台系统拉取，数据加工的逻辑变得非常容易管理，管理员可以非常方便的在后台系统一览所有的实时数据加工逻辑；面对业务需求的变更，只需要在后台系统的页面上操作修改，实时计算集群上的Flink任务可以自动获得新的数据加工逻辑，并实时生效。

4.结局

爱奇艺的大数据采集体系，从最初简单的统计场景起步，不断的经过实时化，规模化，自动化改造，已经发展成一个能支持千万级QPS峰值，每日数据吞吐超过三千亿条的海量数据实时采集系统。随着未来大数据产品的不断多样化，各种大数据应用和AI战略的不断推进，新的数据需求还会推动着这个架构体系继续优化和调整。

在未来的互联网时代，拥有数据的多寡，数据质量的好坏，会直接影响到上层的各种商业智能分析，算法参数调优，AI机器学习等等数据产品的效果。打造一个快速高效，稳定靠谱的数据收集体系，是公司未来发展大数据和AI战略最重要的基础技术保障。