01

统一实时计算平台建设

这是爱奇艺实时计算平台的演变过程。

早期我们支持用户通过脚本和 Jar 包提交流任务，引擎以 Storm 和 Spark 为主。2017 年，我们引入了 Flink，并且意识到 SQL 相比 Jar 包在开发和运维上有着明显优势，于是我们提供了 SQL 开发平台，支持用户通过 SQL 开发流任务。

接下来随着实时业务的爆发式增长，为了支持构建实时数仓，我们上线了低代码开发平台，支持图形化开发作业。今年我们对这些平台进行了系统的整合，和优化设计，建设了统一的实时计算平台 RCP。

实时计算平台在爱奇艺实时数据体系中处于非常重要的一环，它支持用户开发和管理流任务，实现数据的实时摄取、加工、分发。在建设 RCP 平台之前，我们面临这样几个问题：

1. 实时平台多，用户使用成本和服务方维护成本很高。
2. 数据分散在各个平台，无法共享。
3. 规模大，咨询量大，报障多。
4. 任务数量多，版本杂，导致支持用户的成本高。
5. 架构老，难以适应新的技术架构。

基于这样的背景，我们开始建设统一的实时计算平台 RCP。

我们希望通过 RCP 平台达成三个目标：

1. 实现流数据、流任务的统一管理，促进共享，降低成本。
2. 通过优化的设计，更好地帮助用户实现稳定、高效的数据生产。
3. 通过数据湖、流批一体等新技术，进一步提升业务效果。

上图是 RCP 的整体架构，分为平台层、解析引擎、计算框架、调度层、运行层。

平台层用户操作的入口，提供数据表的管理，作业的开发和运维功能；引擎层是作业的解析引擎。计算框架层是 Flink 和 Spark Streaming；调度层我们目前正在进行流批一体化建设，分别有流任务和批任务的调度器，负责任务的提交和状态监控；任务运行层主要在自建集群，少量在公有云上。

平台建设的第一部分工作是作业开发，结合服务用户的经验，我们总结了以下四个痛点：

1. 一部分用户，不熟悉 SQL，他们希望有门槛更低的开发方式。
2. 很多作业中，数据表的字段多，导致 SQL 冗长，难以维护。
3. 开发中需要适配很多不同的版本，解决依赖冲突问题。
4. 作业中有很多 hardcode 的部分，比如数据表的连接信息和配置。

为了解决好这些问题，我们设计了全视角开发模式，让用户从三层不同的视角来看待数据。

1. 第一层，数据流视角。这是最具体的视角，开发者关注底层数据的具体处理逻辑，适合通过底层 API 来实现。
2. 第二层，数据表视角。开发者关注在数据表之间传递数据的逻辑，适合通过 SQL 来处理。
3. 第三层，数据流转视角。开发者更关注上游输入经过怎样的流转之后输出到下游，这里通过数据流程图的方式来描述，非常直接、高效。

下面详细为大家介绍下全视角开发模式。

1. API 开发，用户可以基于底层 API 进行完全定制的开发，然后将 Jar 包提交到平台来运行，我们支持 Flink 和 Spark Streaming 两种框架。
2. SQL 开发，适合熟悉 SQL 的开发者，为了提升开发效率，我们提供了 SQL 编辑器、语法校验、SQL 格式化等工具。
3. DAG 开发，这是门槛最低的方式，用户将数据流的加工逻辑通过流程图的方式来描述，达到了设计即开发的效果。

同样一段逻辑，分别通过 SQL 和 DAG 来开发，在实际生产中，数据表通常有上百个字段，SQL 会比较冗长，难以维护；而通过 DAG 的方式，数据处理流程非常清晰，迭代维护效率高。

全视角开发上线后，使用这三种开发方式的用户都比较多，它实现的效果有以下四点：

1. 降低了开发门槛，连 SQL 语法也不需要深入掌握。
2. 针对不同场景，用户可以选择效率最高的开发方式。
3. 对于 SQL 和 DAG 任务开发，平台提供了一些提升效率的工具，如 SQL 语法校验，格式化等；DAG 中算子的 schema 可以逐级往下传播，不需要用户去手动编辑字段。
4. 所有类型的作业底层对接统一的元数据中心，用户创建的数据表和 UDF 是通用的。不同类型的作业经过解析之后，运行起来也是等效的。

平台建设的第二部分工作是数据源管理，我们实现了一套数据统一集成方案，分为三个模块。

1. Catalog，它是一个持久化的元数据中心，是统一访问数据表和函数的入口。
2. 数据表，它代表各类形态的数据流和数据集，归属于某个项目，使用时通过 Catalog 名，项目名，数据表名三级限定符来访问。
3. Connector，它是访问数据表的具体实现，包含如下功能，一是按指定的数据格式解析数据，比如 json, PB, 另外，适配 hadoop2 和 hadoop3 两大集群版本，适配了 Flink 1.12，1.15 这两个引擎版本，以及各类数据源版本，比如 HBase 等等。

上图是用户在平台上管理数据表的页面，可以看到平台支持用户集中化的管理各类数据表，包括实时队列，KV 库，离线存储等。每个数据表归属于某个项目，所有者负责维护，实现了项目间数据表的权限隔离。其他项目的用户，经过审批后，也能申请访问这些数据表，从而实现共享。

访问数据表的具体实现是在任务提交中完成的，用户上线作业后，平台会解析出作业使用的所有数据表和函数，查询 Catalog，获取数据表的具体信息，然后从文件服务器获取对应的 Connector Jar 和 UDF Jar，和引擎 Jar 一起提交。这个流程有这样三个特点：

1. 对所有类型的任务是共用的，Connector 的代码是完全复用的。
2. 对任务里每个数据表的 Connector 按需加载，灵活装配。
3. 平台统一来完成了不同版本的适配和解决依赖冲突，减轻了用户的开发负担。

平台建设的第三部分工作是任务管理，主要考虑任务的启动、运行、故障和修复这四个阶段的需求。

任务启动时，要能指定消费位置，以及从之前的状态恢复。任务运行时，需要对任务的运行状态进行监控；能便捷查询到运行指标和日志。发生故障时，能及时发现并发出报警通知，最好平台还能进行故障诊断。最后，还能有一些手段能修复或者减轻故障影响。

任务的启停，我们做了如下优化。

任务运行时的状态数据，平台统一进行托管，用户无需关心。停止时会自动触发状态保存，再启动时会尝试从上次的状态中恢复，最大程度避免状态的丢失；任务启动时支持用户指定消费的位点，从而实现灵活消费。

在任务的运行管理中，指标和监控报警是非常重要的一环。

在整体的架构中，指标投递和报警策略主要依赖 prometheus，报警通知依赖爱奇艺内部的报警服务实现。平台支持了丰富的报警策略配置，包括流量的波动；数据源的消费延迟；以及 CPU，内存相关的指标。报警订阅方面支持灵活配置报警级别，通知策略等。另外，这一套架构我们同样适配了 Spark 流任务。

任务日志采集这部分，为了让用户更便捷地查看日志，平台将所有任务的日志进行了采集，通过 Log4j KafkaAppender 实时将任务日志发送到 Kafka，经过解析后，发送到 ES，在 ES 中对任务名等字段进行索引，在任务管理页面上，用户就能方便地检索日志了。

这套流程有这样几个特点：

1. 日志是异步发送的，不会影响任务的正常运行。
2. 日志可查的范围比较大，目前支持查询当前到最近一周的历史日志。
3. 查询分析方便，支持关键词检索；可以集中分析 JobManager 和全部 TaskManager 的日志。
4. 另外，目前我们正在做的一项工作，是对异常日志做自动的分析，帮助用户更快定位问题。

目前 RCP 平台上线了接近一年的时间，已经替代了全部旧的实时平台。有来自各业务团队的近 300 个开发者，他们在 RCP 上构建了 5000 多个实时任务，这些任务总共处理的数据流量峰值达到了 8 千万条每秒，平台日均处理万亿条数据。

02

近实时数据架构

我们公司传统的数仓体系中，数据来源主要是爱奇艺各类 app 等终端的埋点日志以及各个服务的后台日志，经过日志采集服务分别采集到 Kafka 和 hdfs，形成实时和离线两条数据生产线，最后提供给下游应用，这是典型的 Lambda 架构。

主要存在的问题是两套数据生产线开发维护成本高，指标不一致，以及传统实时，离线链路固有的问题。

为了解决这些问题，我们引入了 Iceberg, Flink CDC 等技术，构建了一个近实时的数据通路，我们是这样定义它的：

1. 数据的范围，涵盖分钟级到历史全量数据。
2. 计算上，只需要开发一次，任务能流式运行，也能批式运行。
3. 数据来源上，支持变更数据。

计算方面，我们采用 Flink 作为统一的计算引擎，在 Flink 1.15 版本，已经提供了较为完备的流批统一 API，具备较成熟的批处理能力。

平台侧，RCP 正在支持流批一体化的开发，在开发时能分别配置两种运行模式下 读取数据源的规则，比如批运行时按分区读取数据表，流运行时读取表的新增数据，分别进行批式运行和流式运行。从而实现一次开发，两种方式运行。

在存储上，我们目前以 Iceberg 作为近实时的存储。它主要有三个特点：

1. 实现存量数据加增量数据的统一存储。
2. 支持流式和批式的读写，从而与两种运行模式的计算任务适配。
3. 支持行级更新，从而能导入 MySQL 等数据库的数据。

引入 Iceberg 后，我们做了一些适配工作：对 Iceberg 表进行了平台化管理。包括建表、配置数据的 TTL、文件合并策略等等；支持构建近数据生产 Pipeline，比如分区写入完成后可以生成 done 标记。增量消费时，可以进行延时监控；利用 alluxio 加速 Iceberg 表的查询，在实际业务查询中，起到了比较明显的效果。

接下来是 MySQL 数据接入。很多业务数据在 MySQL 中的，为了对这些数据进行查询分析，一般会把它们同步到大数据系统中。常见的做法会有两个链路，存量数据通过离线方式同步到 Hive，增量数据实时同步到 ES，Kudu 等存储中。

这个方案主要存在以下几个问题：

1. 存量和增量数据在两份存储中，使用不方便。
2. 维护两个同步链路，维护成本较高。
3. 难以保障数据一致性，特别是存量同步切换到增量同步的时候。

经过调研，我们认为 Flink CDC 技术非常适合我们的场景，可以解决刚才提到的问题。主要考虑到它有以下几个优势：

1. 能很好的实现先同步存量数据，再无缝对接到增量同步，且端到端数据一致。
2. Flink CDC2.0 版本之后，实现了无锁同步方案，对源库的影响较小。
3. 支持边同步边数据加工，一个任务实现数据同步、加工、分发，架构简洁。

为了将 Flink CDC 集成到 RCP 平台，我们做了以下工作：将当时 Flink CDC 的版本和 Flink 1.15 做了适配；对 MySQL CDC 类型数据表进行了统一集成，平台对接了 MySQL 服务，打通账号和权限流程，从而规范和简化了用户使用；解决了我们在实践中遇到的数据同步失败的问题。

下面我们对近实时架构做个总结。首先，它适用的场景是对数据时效性和数据分析范围，这两个需求比较均衡的业务。即时效性不要求秒级延迟，同时需要分析较长时间范围的数据，这类业务比较适合。

它相比传统 Lambda 架构的优势主要体现在 ，一套流程带来的开发维护效率提升，以及成本的降低。另外，它能提供时效性和完整性均衡的数据，且能支持接入传统数据库的数据。

同时，也存在一些不足，目前主要是两点：

1. 增加了表维护成本，需要不断地进行文件合并。
2. 存储上提供的能力还是不够全面。比如随机读取能力较弱。

03

业务实践

第一个案例是 BI 普通播放报表近实时通路建设。之前这是一个传统的 Lambda 架构，也遇到了我们刚才谈到的问题。经过和业务同学沟通，了解到这个业务延迟从秒级降级到分钟级是可以接受的，因此我们着手构建了近实时链路，来替代现有的流批两条链路。

在这个链路中，原始数据发送到 Kafka 之后，会保存一份到 hdfs，做故障恢复。然后 ODS 层和 DWD 层都是基于 Iceberg 构建，整个链路是流式运行的。改造完成后的效果主要有 3 点：

1. 整个通路的数据都是流动的，一份存储支持了近实时指标和离线指标的计算。
2. 统一了数据口径，新通路的数据误差与原来的差距在 0.1%以内。
3. 成本显著降低，主要是资源成本和维护成本。

第二个案例是审核业务数据入湖的改造。这个业务的数据架构的审核数据会存到到 mongodb 中，在 ES 里构建二级索引，提供线上查询。旧方案的痛点是，经常会有出统计报表或者批量导出数据的需求，对线上服务构成较大压力。引入数据湖能较好地解决当前问题，原始数据流通过 Kafka，实时同步到 Iceberg 表中，通过 SparkSQL 进行即时分析。达到了以下三个效果：

1. 历史数据可以存在 Iceberg 表中，解除了线上存储的瓶颈。
2. 批量扫描的查询都走 Iceberg，缓解了线上服务的查询压力。
3. 支持即席查询，从而能支持快速统计审核效果，数据批量导出等需求。

第三个是通过 Flink CDC 实现了库存计算业务的改造。整体流程上，业务 MySQL 库中的多张表需要做关联后，结果同步到 Redis 作维度表，实时流再来查询这个维度表。在改造前，是一个定时任务，每隔 10 分钟读取 MySQL 表的全量数据，多张表做关联后，结果写入 Redis，主要存在两个问题。

1. 定时任务有不可避免的调度延迟。
2. 每次读取 MySQL 全表数据再做关联计算，计算量较大，效率比较低。

因此，我们在改造方案中，我们引入了 Flink CDC , 进行一次存量同步后，无缝切换到增量同步，多张表的关联计算的结果写入 Redis，相比旧方案有明显的优势：整个过程是实时的，没有调度延迟，整体延迟从 20 分钟提升到了秒级，因此计算结果的准确性大大提高了；存量同步阶段完成后，后续都是基于增量数据计算，无需重复读取 MySQL 表的全量数据，计算效率显著提升了。

04

未来规划

我们规划了两个大的方向。

1. 第一个方向是平台治理。数据层面，实现数据资产更好的管理，进一步提升数据的共享率；任务层面，平台支持自动排障，减轻用户的运维负担；资源层面，实现计算资源的主动伸缩，更合理利用资源，降低成本。
2. 第二个方向是实现流式数仓。这方面我们跟社区的理念是一致的，希望整个数据通路能实时流动起来，且每个环节的数据都可支持分析，从而实现更高程度的流批统一，为业务创造新的价值。