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

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

　　01 存储实时数仓架构背景

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

　　/ 实时数据数仓链路　　

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

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

　　/ Flink数仓问题与挑战

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

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

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

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

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

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

　　/ 目标与愿景

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

　　● 降低开发门槛，为终极目标。通过降低门槛，提升效率，希望能够达到类似于离线开发一样的效率。同时，解决实时领域复杂的方案设计问题，比如多表JOIN和维度表实时变更。维表实施变更之后，相应的值如何迅速进行更正，这也是一大业务痛点。从而更好地应对创新业务口径经常变更的情况。

　　● 提高开发效率，只需开发SQL，无需关心底层运维设置，实现单一职责化。由于Flink状态中间数据不可查，如何进行更快速更高效的数据测试也非常关键，毕竟不是把数据开发好就够了，还要保障数据的准确性。实时数据的错误，可能会造成主播、电商或平台三方的资损。

　　● 资源成本节省。Flink任务是常驻任务会有大量的资源消耗，我们希望通过架构优化降低资源成本。

　　02 存储实时数仓架构体系

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

　　/ 存储实时数仓架构　　

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

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

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

　　/ 存储实时数仓架构生态　　

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

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

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

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

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

　　/ 一站式研发平台　　

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

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

　　/ 调度引擎挑战

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

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

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

　　● T+0调度支持

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

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

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

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

　　● 实时数据容易晚到

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

　　● 调度间隔

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

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

　　● 首先，支持了T+0参数替换功能，提供了高级的运算法则，可以进行秒级或分钟级的时间偏移。

　　● 其次，对调度引擎进行了深度改造，实现了水平扩展，支持多个scheduler，使得调度引擎可以横向无限扩展。

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

　　● 另外，针对数据容易晚到的问题，我们采取了数据补偿机制，即定时进行数据补偿操作来确保数据的完整性。例如，对于一个分钟级时效的任务，每分钟执行一次后，我们会在数据可能晚到的情况下进行定时补偿，以覆盖完整数据。

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

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

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

　　/ Doris引擎挑战

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

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

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

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

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

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

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

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

　　● 首先，关于多机房容灾能力的问题，在三个机房中每个机房都有一张表的情况下，每张表有三个副本，其中一个副本分摊在一个机房，从生产端的MQ数据写入到Doris后，经过中间加工端再到消费端，最终形成了数据服务的全链路高可用性。在单个机房挂掉时，无论是生产还是消费，都会有同机房优先和跨机房降级策略来保障高效性和稳定性。

　　● 读写隔离机制较为简单，将读写流量分流到不同的集群组上。

　　● 跨集群读写采用两种机制：一种是借助Spark将数据源格式读到Yarn集群，再同步到不同集群；另一种是在Doris内部使用Doris原生能力将集群数据同步到另一个集群。两种方式各有优势，Spark on Doris相对更加稳定且不消耗Doris计算资源，而第二种方案效率更高，根据业务场景和时效性诉求选择不同的跨集群读写方式。

　　03 存储实时数仓架构实践

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

　　/ Flink链路　　

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

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

　　/ 实时榜单解决方案

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

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

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

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

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

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

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

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

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

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

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

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

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

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