抖音集团基于 Apache Doris 的实时数据仓库实践

用户案例
2024/06/13
抖音集团数据 BP 团队 李枭冰

在直播、电商等业务场景中存在着大量实时数据,这些数据对业务发展至关重要。而在处理实时数据时,我们也遇到了诸多挑战,比如实时数据开发门槛高、运维成本高以及资源浪费等。

此外,实时数据处理比离线数据更复杂,需要应对多流 JOIN、维度表变化等技术难题,并确保系统的稳定性和数据的准确性。本文将分享基于 Apache Doris 的实时数仓架构在不同业务场景的实践经验,以及该架构带来的收益。

存储实时数仓架构背景

首先介绍存储实时数仓架构的背景。

01 实时数据数仓链路

实时数据数仓链路.png

目前实时数据主要使用 Flink 作为中转工具,Kafka 作为 Flink 的逻辑表,实现数据在不同数据分层之间的流转。Kafka 本身没有逻辑表,无法像 Hive 那样清晰地进行开发过程。

实时数据和离线数据的内容生产量级会有比较大落差,主要原因在于实时数据开发成本、运维成本以及资源成本,尤其是前两者相较离线开发更高,因此尽管有一部分实时数据的需求,我们经常会想办法将其降级。

  • 开发门槛高:Flink 是有状态的一套数据流引擎,具有状态的增量特性,需要更清晰的底层认知,特别是在多流 JOIN 等场景下。增量的状态,导致无法像 Hive 那样把全量的数据状态存到内存里,进一步进行简单的数据操作。

    实时数据涉及的数据存储量较大,需要使用多种计算引擎,如 OLTP 引擎(MySQL、PostgreSQL)、OLAP 引擎(ClickHouse、Doris)、KV 存储(Abase、Tier、Redis)等,以适应不同的计算需求,这也增加了开发难度。另外,由于其增量状态,也让测试变得困难。

  • 开发运维成本高:复杂的多流 JOIN 操作经常需要存储大量状态数据,这可能会导致稳定性问题,尤其是在处理连续直播等情况下。

    在多个业务线的平台中,一些发展中的业务线由于需要不断进行业务创新,业务口径随之变化,而 Flink 作为增量状态存储的系统,遇到状态不可恢复的问题是不可避免的。当数据口径变更时,直接上线可能会由于状态结构改变而无法进行数据恢复。

  • 资源浪费:在实时场景中,资源浪费是很常见的情况,虽然资源浪费不是核心问题,但是目前各个公司都有治理的需求。

    对于一个任务来说,比如在大促活动刚开始的时候,会有大的潮汐洪峰,但过了几分钟之后,流量会迅速地递减退变,为了保证稳定性,我们需要保持高资源位,来稳定地进行 24x7 的运行,这就会导致资源浪费。

03 目标与愿景

我们希望找到一个架构,能在三个方面做出提升:

  • 降低开发门槛,为终极目标。 通过降低门槛,提升效率,希望能够达到类似于离线开发一样的效率。同时,解决实时领域复杂的方案设计问题,比如多表 JOIN 和维度表实时变更。维表实施变更之后,相应的值如何迅速进行更正,这也是一大业务痛点。从而更好地应对创新业务口径经常变更的情况。
  • 提高开发效率,只需开发 SQL,无需关心底层运维设置,实现单一职责化。 由于 Flink 状态中间数据不可查,如何进行更快速更高效的数据测试也非常关键,毕竟不是把数据开发好就够了,还要保障数据的准确性。实时数据的错误,可能会造成主播、电商或平台三方的资损。
  • 资源成本节省。 Flink 任务是常驻任务会有大量的资源消耗,我们希望通过架构优化降低资源成本。

存储实时数仓架构体系

接下来介绍实时数仓的运转方式。

01 存储实时数仓架构

存储实时数仓架构.png

上图中简明地展示了目前运行架构。

左侧是我们所采用的一套已较为成熟的架构,主要用于一些成熟业务。数据存储方面使用了 Kafka 的逻辑表形式。虽然这种逻辑表缺少字段和约束,并且数据的可查性也不是很好,但却负责了一半以上的实时数据开发。

右侧的架构则更为简单,类似于离线 Hive,采用了 Doris 存储架构。通过 OLAP 引擎和秒级调度,实现了数据分层,可以复用离线开发的内容,使实时数据开发变得更加清晰简洁。整体架构的核心是调度引擎(秒级调度)加上 OLAP 引擎。

02 存储实时数仓架构生态

存储实时数仓架构生态.png

这个架构看似简单,但实际上有着复杂的生态系统在支撑。

这套架构已经运行多年,但仍需要相应的生态系统配合,比如数据质量检查平台和数据质量保障措施。另外,数据治理也是必不可少的,特别是在处理大量数据表、数据模型和数据任务时。

应用数据开发方面,可以通过 Doris 引擎进行数据生产,但如何对外提供数据则需要考虑不同的透出形式。我们通过数仓表直接透出,也可以通过 ETL 数据集成将数据导入到 KV 存储,以满足一些高 QPS 的场景需求。

此外,从数仓模型、数据开发、开发规范到指标体系的建设也是必要的。

这套架构在宏观上与离线系统有类似之处。

03 一站式研发平台

 一站式研发平台.png

我们提供了一站式的数据开发服务。首先是注册数据源,然后通过简单的 SQL 语句即可轻松地进行任务开发。

开发完成后,通过一些配置,实现版本管理、上线、Review、数据回溯、告警、大盘等一系列操作。

04 调度引擎挑战

实时生态系统非常复杂,实践中会遇到一些困难。

实时场景核心有两套引擎:调度引擎和 OLAP 引擎。

调度引擎面临的挑战主要有以下三方面:

4-1: T+0 调度支持

原本我们计划直接复用离线调度引擎,但实际落地时发现了一些问题。比如,离线调度通常是 T+1 的,业务时间的替换可能是不符合准实时开发要求的,准实时或实时开发需要 T+0 的日期参数,一些重跑和依赖调度能力等都需要重新构建。

T+1 离线调度对延时的容忍度较高,稍微延迟几分钟是可以接受的,并且离线调度引擎会采用打散任务的策略来处理这种情况。比如,在 0 点的时候,系统会将一些任务进行打散,部分任务稍晚执行,这在离线环境中非常常见。

但是,在实时场景下,这种延迟是不允许的。另外,实时场景和离线场景的数据量差异很大,实例存储的数据量可能有两、三个数量级以上的差距。

比如天级任务每天只有一个实例,小时级任务有几十个,而分钟级任务则有上千个实例,相差了两个数量级以上了,而秒级任务相差的数量级会更大。这种数据量的差异对存储和调度造成挑战。

4-2: 实时数据容易晚到

因为要处理当天或小时内的数据,而数据的到达可能会有延迟。在这里,类似 Flink 中的 watermark 概念变得非常重要,调度引擎需要支持类似的机制来容忍数据的晚到,并保证数据的完整性。

4-3: 调度间隔

这是一个非常严格的要求,比如 15 秒间隔的任务可能因数据量的关系需要 16 秒完成,这也是需要解决的难题之一。

针对 T+0 调度中的三个难题,我们采取了相应的解决方案

  • 首先,支持了 T+0 参数替换功能,提供了高级的运算法则,可以进行秒级或分钟级的时间偏移。
  • 其次,对调度引擎进行了深度改造,实现了水平扩展,支持多个 scheduler,使得调度引擎可以横向无限扩展。
  • 调度间隔。这是一个非常严格的要求,比如 15 秒间隔的任务可能因数据量的关系需要 16 秒完成,这也是需要解决的难题之一。
  • 另外,针对数据容易晚到的问题,我们采取了数据补偿机制,即定时进行数据补偿操作来确保数据的完整性。例如,对于一个分钟级时效的任务,每分钟执行一次后,我们会在数据可能晚到的情况下进行定时补偿,以覆盖完整数据。

针对任务跑的时间长于调度间隔的问题,我们提出了 MisFire 处理策略,这个策略源自于 Quartz 的一些思想。 针对不同的情况,有多种处理方式。最简单的是任务并行,这也是离线开发的默认方式。

另外一种方式是任务串行,特别适用于实时数据场景,避免数据乱序导致数据不准确。

还有一种方式是数据跳过,如果出现任务积压的情况,系统会自动跳过一些任务实例,以确保任务能够相对健康地运行。比如说,当任务积压了几百个实例时,下一次运行时会将相应的实例 Kill 掉,然后继续运行最新的实例。具体的处理方式需要根据业务场景来确定。

05 Doris 引擎挑战

前面介绍了调度引擎面临的挑战和解决方案,接下来看一下 OLAP 引擎。OLAP 引擎主要面临以下三方面挑战:

  • 跨机房容灾能力:准实时领域跟服务端的一些情况有些类似,即在稳定性方面有着高的要求。一旦出现主播跟播时在线人数突然跳零,就会导致主播的一些话术无法及时组织和应变,进而产生严重的资损。

    因此,我们需要跨机房容灾的能力,来应对单机房故障带来的整体服务不可用,以及实时数据无法对外提供的问题。

  • 读写隔离能力:这涉及到 Doris 平台上的操作。我们同时进行数据的生产和消费,但在数据最初阶段,缺乏有效的隔离措施,而这对数据的稳定性是至关重要的。

  • 跨集群 ETL 能力:我们对不同业务场景有着严格的重要等级要求,会将数据分散到多个集群中,比如 A 业务集群、B 业务集群和 C 业务集群等。

B 或 C 都是交易类的依赖订单流的数据,会有公共数仓的建设,这些公共数仓的建设如果无法实现从 B 集群同步到 C 集群,就会导致不同业务线或集群之间的重复建设,无论从人力还是资源方面都会给我们带来负担。

特别是对于涉及交易类数据的集群,这种同步工作显得尤为重要。因此,跨集群 ETL 是我们数仓建设中非常核心的一个能力。

 Doris 引擎挑战.png

针对上述问题,一一进行解决。

  • 首先,关于多机房容灾能力的问题,在三个机房中每个机房都有一张表的情况下,每张表有三个副本,其中一个副本分摊在一个机房,从生产端的 MQ 数据写入到 Doris 后,经过中间加工端再到消费端,最终形成了数据服务的全链路高可用性。在单个机房挂掉时,无论是生产还是消费,都会有同机房优先和跨机房降级策略来保障高效性和稳定性。
  • 读写隔离机制较为简单,将读写流量分流到不同的集群组上。
  • 跨集群读写采用两种机制:一种是借助 Spark 将数据源格式读到 Yarn 集群,再同步到不同集群;另一种是在 Doris 内部使用 Doris 原生能力将集群数据同步到另一个集群。两种方式各有优势,Spark on Doris 相对更加稳定且不消耗 Doris 计算资源,而第二种方案效率更高,根据业务场景和时效性诉求选择不同的跨集群读写方式。

存储实时数仓架构实践

接下来简要介绍一些实际的应用场景。

存储实时数仓架构实践-Flink.png

Flink 链路如上图所示,第一条链路看起来比较复杂,需要执行多条流的 JOIN 操作。

使用基于存储的实时数仓架构后,整体结构变得更加简洁,虽然数据来源仍为多条流,但实际上在一张表里进行了 JOIN 操作。整体涉及了四五个甚至更多流式 JOIN,流式 JOIN 复杂度大家都比较了解。不过,实际负责的 JOIN 可能仅有三个。开发成本和后期维护成本都大幅降低。

02 实时榜单解决方案

另一个是实时榜单解决方案。

实时榜单解决方案.png

针对这种场景,我们进行了解决方案的抽象,并在存储数仓中实施了一个方案。

最初的方案是基于 Flink 的,出现了一些问题,于是后期迁移到了基于 Doris 的存储数仓方案。这套方案的特点是元数据定义比较清晰。

元数据由实时表从 MQ 中的字段解析而来,解析后对其进行了一些元数据定义,即对榜单场景业务逻辑进行抽象,比如会定义周期、原子指标以及如何加工这些原子指标。

另外,还定义了榜单如何进行分区,分区可以根据实体类型来确定,例如对商家、视频或直播进行排名。通过简单的配置,能够快速创建出相应的 Flink 任务。

在业务实际运营中,有许多类似的榜单场景,这样的榜单场景过多导致出现了两个问题。

首先,榜单场景过多导致任务量激增,这会给资源治理带来较多困难。特别是对于实时流处理,需要 24 小时全天候运行,任务量增加会让资源治理问题变得更加严峻。

其次,报警运维也是一个挑战,实时任务报警频率高,甚至一个任务可能随时都会产生警报。而随着任务数量的增加,报警更加频繁。此外,由于大量任务消费同一个消息队列,会放大流量,给 HDFS 带来额外负担。

另外,电商领域的大型促销活动常常伴随着长周期状态,这种长周期计算会对 Flink 的大状态稳定性产生影响,同时也使回溯变得困难。为应对这些问题,运维人员经常需要在零点进行操作,只有在这个时间点才重新计算,相对来说状态比较小,回溯压力也比较小。

实时榜单解决方案-2.png

基于上述痛点,我们将 Flink 架构迁移到了存储数仓架构,使得运维工作变得更加高效。相比 Flink,在榜单场景下资源量和报警量都有下降。并且解决了长周期计算的难题。由于状态保存在 Doris 的表中,长周期计算变得更加灵活。

存储实时数仓架构规划

最后分享我们在未来要做的一些工作。

首先是对解决方案的封装。我们已经封装了一个榜单业务场景,还有许多其他场景,比如 DMP、标签和中间层数据等,这些场景都可以被打包成解决方案。除了模式和方法论的封装之外,还有存储架构的封装。

在存储架构方面,不断演进自研的数据湖产品,扩展更多的存储架构。

另外是智能化运维整合,实时数据的稳定性对开发和运维人员压力是非常大的,我们希望整合一些规则和算法,实现自动化处理部分场景,剩下的做推荐化预案,从而提升 MTTR,提升故障恢复的时效性并降低成本。

以上就是本次分享的内容,谢谢大家。