Flink 在爱奇艺广告业务的实践

简介： 5 月 22 日北京站 Flink Meetup 分享的议题。

本文整理自爱奇艺技术经理韩红根在 5 月 22 日北京站 Flink Meetup 分享的议题《Flink 在爱奇艺广告业务的实践》，内容包括：

1. 业务场景
2. 业务实践
3. Flink 使用过程当中的问题及解决
4. 将来规划

1、业务场景

实时数据在广告业务的使用场景主要能够分为四个方面：

• 数据大屏：包括曝光、点击、收入等核心指标的展现，以及故障率等监控指标；
• 异常监测：由于广告投放的链路比较⻓，因此若是链路上发生任何波动的话，都会对总体的投放效果产生影响。除此以外，各个团队在上线过程当中是否会对总体投放产生影响，都是经过异常监测系统可以观测到的。咱们还可以观测业务指标走势是否合理，好比在库存正常的状况下，曝光是否有不一样的波动状况，这能够用来实 时发现问题;
• 数据分析：主要用于数据赋能业务发展。咱们能够实时分析广告投放过程当中的一些异常问题，或者基于当前的投放效果去研究怎样优化，从而达到更好的效果;
• 特征工程：广告算法团队主要是作一些模型训练，用于支持线上投放。技术特征最初大部分是离线，随着实时的发展，开始把一些工程转到实时。

2、业务实践

业务实践主要分为两类，第一个是实时数仓，第二个是特征工程。

1. 实时数仓

1.1 实时数仓 - 目标

实时数仓的目标包括数据完整性、服务稳定性和查询能力。

• 数据完整性：在广告业务里，实时数据主要是用于指导决策，好比广告主须要根据当前投放的实时数据，指导后面的出价或调整预算。另外，故障率的监控须要数据自己是稳定的。若是数据是波动的，指导意义就很是差，甚至没有什么指导意义。所以完整性自己是对时效性和完整性之间作了一个权衡；
• 服务稳定性：生产链包括数据接入、计算（多层）、数据写入、进度服务和查询服务。除此以外还有数据质量，包括数据的准确性以及数据趋势是否符合预期；
• 查询能力：在广告业务有多种使用场景，在不一样场景里可能使用了不一样的 OLAP 引擎，因此查询方式和性能的要求不一致。另外，在作数据分析的时候，除了最新最稳定的实时数据以外，同时也会实时 + 离线作分析查询，此外还包括数据跨源和查询性能等要求。

1.2 实时数仓 - 挑战

• 数据进度服务：须要在时效性和完整性之间作一个权衡。
• 数据稳定性：因为生产链路比较长，中间可能会用到多种功能组件，因此端到端的服务稳定性对总体数据准确性的影响是比较关键的。
• 查询性能：主要包括 OLAP 分析能力。在实际场景中，数据表包含了离线和实时，单表规模达上百列，行数也是很是大的。

1.3 广告数据平台架构

上图为广告数据平台基础架构图，从下往上看：

• 底部是数据采集层，这里与大部分公司基本一致。业务数据库主要包含了广告主的下单数据以及投放的策略；埋点日志和计费日志是广告投放链路过程当中产生的日志；

• 中间是数据生产的部分，数据生产的底层是大数据的基础设施，这部分由公司的一个云平台团队提供，其中包含 Spark / Flink 计算引擎，Babel 统一的管理平台。Talos 是实时数仓服务，RAP 和 OLAP 对应不一样的实时分析以及 OLAP 存储和查询服务。

  数据生产的中间层是广告团队包含的一些服务，例如在生产里比较典型的离线计算和实时计算。

  • 离线是比较常见的一个分层模型，调度系统是对生产出的离线任务作有效的管理和调度。
  • 实时计算这边使用的引擎也比较多，咱们的实时化是从 2016 年开始，当时选的是 Spark Streaming，后面随着大数据技术发展以及公司业务需求产生了不一样场景，又引入了计算引擎 Flink。
  • 实时计算底层调度依赖于云计算的 Babel 系统，除了计算以外还会伴随数据治理，包括进度管理，就是指实时计算里一个数据报表当前已经稳定的进度到哪一个时间点。离线里其实就对应一个表，有哪些分区。
  • 血缘管理包括两方面，离线包括表级别的血缘以及字段血缘。实时主要仍是在任务层面的血缘。
  • 至于生命周期管理，在离线的一个数仓里，它的计算是持续迭代的。可是数据保留时间很是长的话，数据量对于底层的存储压力就会比较大。
  • 数据生命周期管理主要是根据业务需求和存储成本之间作一个权衡。
  • 质量管理主要包括两方面，一部分在数据接入层，判断数据自己是否合理；另一部分在数据出口，就是结果指标这一层。由于咱们的数据会供给其余不少团队使用，所以在数据出口这一层要保证数据计算没有问题。

• 再上层是统一查询服务，咱们会封装不少接口进行查询。

  • 由于数据化包括离线和实时，另外还有跨集群，因此在智能路由这里会进行一些选集群、选表以及复杂查询、拆分等核心功能。
  • 查询服务会对历史查询进行热度的统一管理。这样一方面能够更应进一步服务生命周期管理，另外一方面能够去看哪些数据对于业务的意义很是大。
  • 除了生命周期管理以外，它还能够指导咱们的调度系统，好比哪些报表比较关键，在资源紧张的时候就能够优先调度这些任务。
• 再往上是数据应用，包括报表系统、Add - hoc 查询、数据可视化、异常监控和下游团队。

1.4 实时数仓 - 生产链路

数据生产链路是从时间粒度来说的，咱们最开始是离线数仓链路，在最底层的这一行，随着实时化需求推动，就产生了一个实时链路，整理来讲，是一个典型的 Lambda 架构。

另外，咱们的一些核心指标，好比计费指标，由于它的稳定性对下游比较关键，因此咱们这边采用异路多活。异路多活是源端日志产生以后，在计算层和下游存储层作了彻底的冗余，在后面的查询里作统一处理。

1.5 实时数仓 - 进度服务

上文介绍了咱们要求提供出去的实时数据的指标是稳定不变的，进度服务实现的核心点包括时间窗口里指标的变化趋势，同时结合了实时计算任务自己的状态，由于在实时数仓里，不少指标是基于时间窗口作聚合计算。

好比一个实时指标，咱们输出的指标是 3 分钟，也就是说 4：00 这个时间点的指标的就包括了 4：00～4：03 的数据，4：03 包括了 4：03～4：06 的数据，其实就是指一个时间窗口的数据，何时是对外可见的。由于在实时计算里，数据不断进来， 4：00 的时间窗口的数据从 4：00 开始，指标就已经开始产生了。随着时间叠加，指标不断上升，最后趋于稳定。咱们基于时间窗口指标的变化率，来判断它是否趋于稳定。

但若是只是基于这个点来看，那么它还存在必定的弊端。

由于这个结果表的计算链会依赖不少个计算任务，若是这个链路上面哪一个任务出现问题，可能会致使当前的指标虽然走势已经趋于正常，可是最终并不完整。因此在这基础之上，咱们又引入了实时计算任务状态，在指标趋于稳定的时候，同时去看生产链路上这些计算任务是否正常，若是是正常的话，表示任务自己时间点的指标已经稳定，能够对外提供服务。

若是计算有卡顿、堆积，或者已经有异常在重启过程当中，就须要继续等待迭代处理。

1.6 实时数仓 - 查询服务

上图为查询服务架构图。

最下方是数据，里面有实时存储引擎，包括 Druid 等。在离线中，数据在 Hive 里边，可是在作查询的时候，会把它们进行 OLAP 的同步，在这边使用的是两种引擎。为了和 Kudu 作 union 查询，会把它同步到 OLAP 引擎，而后上面去统一使用 Impala 作查询。另外，对于使用场景里比较固定的方式，能够导到 Kylin 里，而后在上面作数据分析。

基于这些数据，会有多个查询节点，再上面是一个智能路由层。从最上面查询网关，当有一个查询请求进来，首先判断它是否是一个复杂场景。好比在一个查询里，若是它的时长同时跨越了离线和实时，这里就会同时使用到离线表和实时表。

另外，离线表里还有更复杂的选表逻辑，好比小时级别，天级别。通过复杂场景分析以后，就会把最终选择的表大概肯定下来。其实在作智能路由的时候，才会去参考左边的一些基础服务，好比元数据管理，当前这些表的进度到哪一个点了。

对于查询性能的优化，在数据里，底层扫描的数据量对最终性能的影响是很是大的。因此会有一个报表降维，根据历史的查询去作分析。好比在一个降维表包含哪些维度，能够覆盖到百分之多少的查询。

1.7 数据生产 - 规划

以前在实时数据报表生产里提到，它主要是基于 API 的方式实现的。Lambda 架构自己有一个问题就是实时跟离线是两个计算团队，对于同一个需求，须要两个团队同时去开发，这样会带来几个问题。

• 一方面是他们的逻辑可能会发生差别，最终致使结果表不一致；
• 另外一方面是人力成本，同时须要两个团队进行开发。

所以咱们的诉求是流批一体，思考在计算层是否可使用一个逻辑来表示同一个业务需求，好比能够同时使用流或者批的计算引擎来达到计算的效果。

在这个链路里边，原始数据经过 Kafka 的方式接入进来，通过统一的 ETL 逻辑，接着把数据放在数据湖里。由于数据湖自己能够同时支持流和批的方式进行读写，并且数据湖自己能够实时消费，因此它既能够作实时计算，也能够作离线计算，而后统一把数据再写回数据湖。

前文提到在作查询的时候，会使用离线跟实时作统一整合，因此在数据湖里写同一个表，在存储层面能够省去不少工做，另外也能够节省存储空间。

1.8 数据生产 - SQL 化

SQL 化是 Talos 实时数仓平台提供的能力。

从页面上来看，它包括了几个功能，左边是项目管理，右边包括 Source、Transform 和 Sink。

• 有一些业务团队自己对于计算引擎算子很是熟，那么他们即可以作一些代码开发；
• 可是不少业务团队可能对引擎并非那么了解，或者没有强烈的意愿去了解，他们就能够经过这种可视化的方式，拼接出一个做业。

例如，能够拖一个 Kafka 的数据源进来，在上面作数据过滤，而后就能够拖一个 Filter 算子达到过滤逻辑，后面能够再去作一些 Project，Union 的计算，最后输出到某个地方就能够了。

对于能力稍微高一些的同窗，能够去作一些更高层面的计算。这里也能够实现到实时数仓的目的，在里面建立一些数据源，而后经过 SQL 的方式，把逻辑表示出来，最终把这个数据输出到某种存储。

上面是从开发层面来说，在系统层面上，它其实还提供了一些其余的功能，好比规则校验，还有开发/测试/上线，在这里能够统一管理。此外还有监控，对线上跑的实时任务有不少实时指标，能够经过查看这些指标来判断当前的任务是否是正常的状态。

2. 特征工程

特征工程有两方面的需求：

• 第一个需求是实时化，由于数据价值随着时间的递增会愈来愈低。好比某用户表现出来的观影行为是喜欢看儿童内容，平台就会推荐儿童相关的广告。另外，用户在看广告过程当中，会有一些正/负反馈的行为，若是把这些数据实时迭代到特征里，就能够有效提高后续的转化效果。

  实时化的另外一个重点是准确性，以前不少特征工程是离线的，在生产环节里面存在计算时的数据跟投放过程当中的特征有误差，基础特征数据不是很准确，所以咱们要求数据要更实时、更准确。

• 特征工程的第二个需求是服务稳定性。

  • 首先是做业容错，好比做业在异常的时候可否正常恢复；
  • 另外是数据质量，在实时数据里追求端到端精确一次。

2.1 点击率预估

下面是在特征实时化里的实践，首先是点击率预估的需求。

点击率预估案例的背景如上所示，从投放链路上来讲，在广告前端用户产生观影行为，前端会向广告引擎请求广告，而后广告引擎在作广告召回粗排/精排的时候会拿到用户特征和广告特征。把广告返回给前端以后，后续用户行为可能产生曝光、点击等行为事件，在作点击率预估的时候，须要把前面请求阶段的特征跟后续用户行为流里的曝光和点击关联起来，造成一个 Session 数据，这就是咱们的数据需求。

落实到具体实践的话包括两方面：

• 一方面是 Tracking 流里曝光、点击事件的关联；
• 另外一方面是特征流跟用户行为的关联。

在实践过程当中有哪些挑战？

• 第一个挑战是数据量；
• 第二个挑战是实时数据乱序和延迟；
• 第三个挑战是精确性要求高。

在时序上来讲，特征确定是早于 Tracking，可是两个流成功关联率在 99% 以上的时候，这个特征须要保留多久？由于在广告业务中，用户能够离线下载一个内容，在下载的时候就已经完成了广告请求和返回了。可是后续若是用户在没有网的状况下观看，这个事件并不会立马返回，只有当状态恢复的时候，才会有后续曝光和点击事件回传。

因此这个时候，其实特征流和 Tracking 的时间归纳是很是长的。咱们通过离线的数据分析，若是两个流的关联率达 99% 以上，那么特征数据就须要保留比较长的时间，目前是保留 7 天，这个量级仍是比较大的。

上图为点击率预测的总体架构，刚才咱们提到关联包括两部分：

• 第一个部分是用户行为流里曝光跟点击事件的关联，这里经过 CEP 实现。
• 第二个部分是两个流的关联，前面介绍特征须要保留 7 天，它的状态较大，已是上百 TB。这个量级在内存里作管理，对数据稳定性有比较大的影响，因此咱们把特征数据放在一个外部存储 (Hbase) 里，而后和 HBase 特征作一个实时数据查询，就能够达到这样一个效果。

可是由于两个流的时序自己多是错开的，就是说，当曝光、点击出现的时候，可能这个特征尚未到，那么就拿不到这个特征。因此咱们作了一个多级重试队列，保证最终两个流关联的完整性。

2.2 点击率预估 - 流内事件关联

上图右边是更细的讲解，阐述了流内事件关联为何选择 CEP 方案。业务需求是把用户行为流里属于同一次广告请求，而且是同一个广告的曝光跟点击关联起来。曝光以后，好比 5 分钟以内产生点击，做为一个正样本，5 分钟以后出现的点击则抛弃不要了。

能够想象一下，当遇到这样的场景，经过什么样的方案能够实现这样的效果。其实在一个流里多个事件的处理，能够用窗口来实现。但窗口的问题是：

• 若是事件序列自己都在同一个窗口以内，数据没有问题；
• 可是当事件序列跨窗口的时候，是达不到正常关联效果的。

因此当时通过不少技术调研后，发现 Flink 里的 CEP 能够实现这样的效果，用相似政策匹配的方式，描述这些序列须要知足哪些匹配方式。另外它能够指定一个时间窗口，好比曝光和点击间隔 15 分钟。

上图左边是匹配规则的描述，begin 里定义一个曝光，实现曝光以后 5 分钟以内的点击，后面是描述一个能够出现屡次的点击，within 表示关联窗口是多长时间。

在生产实践过程当中，这个方案大部分状况下能够关联上，可是在作数据对比的时候，才发现存在某些曝光点击没有正常关联到。

通过数据分析，发现这些数据自己的特色是曝光跟点击的时间戳都是毫秒级别，当它们有相同毫秒时间戳的时候，这个事件就不能正常匹配。因而咱们采用一个方案，人为地对于点击事件加一毫秒，进行人工错位，这样就保证曝光跟点击可以成功关联上。

2.3 点击率预估-双流关联

前文提到特征数据须要保留 7 天，因此状态是上百 TB。须要把数据放在一个外部存储里，所以在作技术选型时对外部存储有必定的要求：

• 首先支持比较高的读写并发能力；
• 另外它的时效性须要很是低；
• 同时由于数据要保留 7 天，因此它最好具有生命周期管理能力。

基于以上几个点，最终选择了 HBase，造成上图的解决方案。

上面一行表示经过 CEP 以后把曝光点击序列关联在一块儿，最下面是把特征流经过 Flink 写到 HBase 里，去作外部状态存储，中间核心模块是用于达到两个流的关联。拿到曝光点击关联以后去查 HBase 数据，若是可以正常查到，就会把它输出到一个正常结果流里。而对于那些不能构成关联的数据，作了一个多级重试队列，在屡次重试的时候会产生队列降级，而且在重试的时候为了减轻对 HBase 的扫描压力，重试 Gap 会逐级增长。

另外还有一个退出机制，由于重试不是无限进行的。退出机制的存在缘由主要包括两个点：

• 第一点是特征数据保留了 7 天，若是对应特征是在 7 天以前，那么它自己是关联不到的。
• 另外在广告业务里，存在一些外部的刷量行为，好比刷曝光或刷点击，但它自己并无真实存在的广告请求，因此这种场景也拿不到对应特征。

所以，退出机制意味着在重试屡次以后就会过时，而后会到重试过时的数据里。

2.4 有效点击

在有效点击场景里，其实也是两个流的关联，可是两个场景里的技术选型是彻底不同的。

首先看一下项目背景，在网大场景里，影片自己就是一个广告。用户在点击以后，就会进入到一个播放页面。在播放页面里，用户能够免费观看 6 分钟，6 分钟以后想要继续观看，须要是会员或者购买才行，在这里须要统计的数据是有效点击，定义是在点击以后观影时长超过 6 分钟便可。

这种场景落实到技术上是两个流的关联，包括了点击流和播放心跳流。

• 点击流比较好理解，包括用户的曝光和点击等行为，从里面筛选点击事件便可。
• 播放行为流是在用户观看的过程，会定时地把心跳信息回传，好比三秒钟回传一个心跳，代表用户在持续观看。在定义时长超过 6 分钟的时候，须要把这个状态自己作一些处理，才能知足 6 分钟的条件。

在这个场景里，两个流动 Gap 相对比较小，而在电影里时长通常是两个多小时，因此点击以后的行为，Gap 基本是在三个小时之内才能完成，所以这里自己的状态是相对比较小的，使用 Flink 的状态管理能够达到这样的效果。

接下来咱们看一个具体的方案。

从流上来看，绿色部分是点击流，蓝色部分是播放心跳流。

• 在左边的状态里面，一个点击事件进来以后，会对这个点击作一个状态记录，同时会注册一个定时器作按期清理，定时器是三个小时。由于大部分影片的时长在三小时之内，若是这个时候对应的播放事件尚未一个目标状态，点击事件基本就能够过时了。

• 在右边的播放心跳流里，这个状态是对时长作累计，它自己是一个心跳流，好比每三秒传一个心跳过来。咱们须要在这里作一个计算，看它累计播放时长是否是达到 6 分钟了，另外也看当前记录是否是到了 6 分钟。对应 Flink 里的一个实现就是把两个流经过 Connect 算子关系在一块儿，而后能够制定一个 CoProcessFunction，在这里面有两个核心算子。

  • 第一个算子是拿到状态 1 的流事件以后，须要作一些什么样的处理；
  • 第二个算子是拿到第 2 个流事件以后，能够自定义哪些功能。

  算子给用户提供了不少灵活性，用户能够在里面作不少逻辑控制。相比不少的 Input Join，用户可发挥的空间比较大。

2.5 特征工程 - 小结

针对以上案例作一个小结。如今双流管理已经很是广泛，有许多方案能够选择，好比 Window join，Interval join，还有咱们使用的 Connect + CoProcessFunction。除此以外，还有一些用户自定义的方案。

在选型的时候，建议从业务出发，去作对应的技术选型。首先要思考多个流之间的事件关系，而后判断出状态是什么规模，必定程度上能够从上面不少方案里排除不可行的方案。

3、Flink 使用过程当中的问题及解决

1. 容错

在 Flink 内部主要是经过 Checkpoint 作容错，Checkpoint 自己是对于 Job 内部的 Task 级别的容错，可是当 Job 主动或异常重启时，状态没法从历史状态恢复。

所以咱们这边作了一个小的改进，就是一个做业在启动的时候，它也会去 Checkpoint 里把最后一次成功的历史状态拿到，而后作初始化管理，这样就达到状态恢复的效果。

2. 数据质量

Flink 自己实现端到端精确一次，首先须要开启 Checkpoint 功能，而且在 Checkpoint 里指定精确一次的语义。另外，若是在下游好比 Sink 端，它自己支持事务，就能够结合两阶段提交与 Checkpoint 以及下游的事务作联动，达到端到端精确一次。

在上图右边就是描述了这个过程。这是一个预提交的过程，就是 Checkpoint 协调器在作 Checkpoint 的时候，会往 Source 端注入一些 Barrier 数据，每一个 Source 拿到 Barrier 以后会作状态存储，而后把完成状态反馈给协调器。这样每一个算子拿到 Barrier，实际上是作相同的一个功能。

到 Sink 端以后，它会在 Kafka 里提交一个预提交标记，后面主要是 Kafka 自己事务机制来保证的。在全部的算子都完成 Checkpoint 以后，协调器会给全部的算子发一个 ACK，发送一个确认状态，这时候 Sink 端作一个提交动做就能够了。

3. Sink Kafka

在以前的实践中咱们发现，下游 Kafka 增长分区数时，新增分区无数据写入。

原理是 FlinkKafkaProducer 默认使用 FlinkFixedPartitioner，每一个 Task 只会发送到下游对应的一个 Partition 中，若是下游 Kafka 的 Topic 的 Partition 大于当前任务的并行度，就会出现该问题。

解决办法有两个：

• 第一个办法是用户自定义一个 FlinkKafkaPartitioner；
• 另外一个办法是默认不配置，默认轮询写入各个 Partition。

4. 监控增强

对于运行中的 Flink 做业，咱们须要查看它自己的一些状态。好比在 Flink UI 里面，它的不少指标都是在 Task 粒度，没有总体的效果。

平台这边对这些指标作了进一步的聚合，统一在一个页面里面展现。

从上图能够看到，展现信息包括反压状态，时延状况以及运行过程当中 JobManager 和 TaskManage 的 CPU / 内存的利用率。另外还有 Checkpoint 的监控，好比它是否超时，最近是否有 Checkpoint 已经失败了，后面咱们会针对这些监控指标作一些报警通知。

5. 监控报警

当实时任务运营异常的时候，用户是须要及时知道这个状态的，如上图所示，有一些报警项，包括报警订阅人、报警级别，下面还有一些指标，根据前面设置的指标值，若是知足这些报警策略规则，就会给报警订阅人推送报警，报警方式包括邮件、电话以及内部通信工具，从而实现任务异常状态通知。

经过这种方式，当任务异常的时候，用户能够及时知晓这个状态，而后进行人为干预。

6. 实时数据生产

最后总结一下爱奇艺广告业务在实时链路生产上面的关键节点。

• 咱们的实时是从 2016 年开始起步，当时主要功能点是作一些指标实时化，使用的是 SparkStreaming；
• 2018 年上线了点击率实时特征；
• 2019 年上线了 Flink 的端到端精确到一次和监控强化。
• 2020 年上线了有效点击实时特征；
• 同年10月，逐步推动实时数仓的改进，把 API 生产方式逐渐 SQL 化；
• 2021 年 4 月，进行流批一体的探索，目前先把流批一体放在 ETL 实现。

以前咱们的 ETL 实时跟离线是分别作的，经过批处理的方式，而后换到 Hive 表里边，后面跟的是离线数仓。在实时里，通过实时 ETL，放到 Kafka 里边，而后去作后续的实时数仓。

先在 ETL 作流批一体的第一个好处是离线数仓时效性提高，由于数据须要作反做弊，因此咱们给广告算法提供基础特征的时候，反做弊以后的时效性对于后续总体效果的提高是比较大的，因此若是把 ETL 作成统一实时化以后，对于后续的指导意义很是大。

ETL 作到流批一体以后，咱们会把数据放在数据湖里面，后续离线数仓和实时数仓均可以基于数据湖实现。流批一体能够分为两个阶段，第一阶段是先把 ETL 作到一体，另外报表端也能够放在数据湖里边，这样咱们的查询服务能够作到一个更新的量级。由于以前须要离线表跟实时表作一个 Union 的计算，在数据湖里面，咱们经过离线和实时写一个表就能够实现了。

4、将来规划

关于将来规划：

首先是流批一体，这里包括两个方面：

• 第一个是 ETL 一体，目前已是基本达到可线上的状态。
• 第二个是实时报表 SQL 化和数据湖的结合。

另外，如今的反做弊主要是经过离线的方式实现，后面可能会把一些线上的反做弊模型转成实时化，把风险降到最低。

原文连接

本文为阿里云原创内容，未经容许不得转载。