Doris实战——拈花云科的数据中台实践

前言

拈花云科 NearFar X Lab 团队调研并引进 Doris 作为新架构下的数据仓库选型方案。本文主要介绍了拈花云科数据中台架构从 1.0 到 2.0 的演变过程，以及 Doris 在交付型项目和 SaaS 产品中的应用实践。

一、业务背景

拈花云科的服务对象主要是国内各个景区、景点，业务范围涵盖文旅行业的多个板块，如票务、交通、零售、住宿、餐饮、影院、KTV、租赁等。多业务线下用户对于数据使用的时效性需求差异较大，需要我们能够提供实时、准实时、T+1的业务支撑能力。同时由于大部分景区为国有化的特点，也需要具备能够提供私有化交付部署及SaaS 化数据中台产品解决方案的双重服务支撑能力。

二、数据中台1.0—Lambda

早期构建数据中台时，为了优先满足B端用户数据整合的需求，以稳定数据输出为出发点，因此我们基于行业中比较成熟的Lambda架构形成了数据中台 1.0 。

在数据中台1.0架构中分为三层，分别为Batch Layer，Speed Layer 和 Serving Layer。其中，Batch Layer用于批量处理全部的数据，Speed Layer用于处理增量的数据，在Serving Layer中综合Batch Layer生成的Batch Views 和 Speed Layer 生成的Realtime Views，提供给用户查询最终的结果。

Batch Layer：在我们早期的实施类项目中，单纯以离线T+1进行数据支持的项目占了绝大多数，但实施类项目在实现 Lambda 架构的过程中也会面临很多问题。例如：由于项目本身的原因，业务系统不能开放DB 的Binlog 供数据仓库采集，因此只能以JDBC的方式完成增量或全量的数据同步，而通过该方式同步的数据，往往会由于源端系统人工补充数据、时间戳不规范等问题产生同步数据差异的情况发生，最终只能通过额外的数据对比逻辑进行校验，以保证其数据的一致性。

Speed Layer：项目受成本约束较大，大面积基于流的实时计算对于不论是从硬件成本、部署成本还是实施成本，维护成本等角度均难以支撑。基于该原因，在实施类项目中只有部分业务会进行基于流的实时统计计算，同时满足流计算条件的业务上游系统也同时满足同步Binlog 的使用需求。

Serving Layer：大部分的预计算结果存储在MySQL 中提供 Report 支持，部分实时场景通过Merge Query 对外提供Ad-Hoc(数据探索)的查询支持。

随着时间的推移，大量的项目交付使用增多，架构的问题也逐渐开始显现：

开发和维度成本高：该架构需要维护两套代码，即批处理和实时处理的代码，这无疑增加了开发和维护的成本。
数据处理复杂度高：Lambda 架构需要处理多个层次的数据，包括原始数据、批处理数据和实时数据，需要对不同的数据进行清洗、转换和合并，数据处理的复杂度较高。
实时计算支持有限：业务方对于数据时效性要求越来越高，但是该架构能力有限，无法支持更多、更高要求的的实时计算。
资源利用率低：离线资源较多，但我们仅在凌晨后的调度时间范围内使用，资源利用率不高。
受成本制约：该架构对于部分用户而言使用成本较高，难以起到降低成本提高效率的作用。

三、新架构的设计目标

基于以上架构问题，我们希望实现一套更加灵活的架构方案，同时希望新的架构可以满足日益增高的数据时效性要求。在新方案实现之前，我们先对当前的业务应用场景和项目类型进行分析。

我们业务应用场景分为以下四类，这四类场景的特点和需求分别是：

看板类Reporting：包括 Web/移动端数据看板和大屏可视化，用于展示景区重要场所的数据，如业务播报（实时在园人数监控、车船调度管理等）、应急管理监控（客流密度监控、景区消防预警、景区能耗监控等）。其组成特点一般是业务汇总指标和监控指标报警，对数据时效性要求较高。
报表类：数据报表以图表形式展示，主要服务于各业务部门的一线业务人员。会更多关注垂直业务的数据覆盖程度，会有钻取需求（也可能通过不同报表来体现不同数据粒度）。一般以景区的业务部门为单位构建报表栏目和分析主题，除财务结算类报表外，一般可接受 T+1 的报表时效。
分析类：自助分析基于较好的数据模型表（数据宽表）实现，对分析人员有一定的数据理解和操作需求，基于我们提供的BI分析平台，业务人员可基于此数据范围通过拖拽的方式组合出自己的数据结果，灵活度较高。该场景对数据时效性要求不高，更多关注业务数据沉淀和与往期历史数据的对比分析，侧重架构的OLAP能力。
服务类：一般对接三方系统，由数据中台提供数据计算的结果。如画像标签等数据，通过数据接口控制权限，提供对外数据服务与其他业务系统集成，需要新架构能够提供稳定的数据服务

接着我们对项目类型的特点和需求也进行了分析，并确定新架构需要同时提供实施类项目和 SaaS 产品的数据中台支撑能力：

四、数据中台2.0—Apache Doris

结合以上需求，我i们计划对原有架构进行升级，并对新架构的OLAP引擎进行选型。在对比了ClickHouse 等OLAP引擎后（社区有非常多的对比文章参考，这里不过多赘述）最终选择了 Apache Doris 作为数据中台 2.0 的基座。同时，在数据的同步、集成即计算环节，我们也构建了多套方案来适配以Doris 为核心的计算链路，以应对不同类型的实施类项目及SaaS产品需求。

数据中台 2.0的核心思路是将Doris作为核心的数据仓库，并将其作为实时数据同步中心，核心数据计算中心。数据集成环节将专注于数据同步，而计算交由Doris 完成或由 Doris 辅助计算引擎完成。同时，我们将在提供多种数据同步至Doris 的方案以应对不同的项目需求。在这个架构下，我们支持实现实时、准实时、T+1的计算场景支持，以满足不同业务场景的需求。

4.1 新架构数据流转

1）数据同步集成：架构 2.0 有多种数据同步方式，我们主要借助Doris Unique Key模型完成数据的同步更新

2）数据分层计算：根据项目资源情况分View/实体表单来构建后面的数据层级（DWD，DWS，ADS）。业务较轻或时效性很高的，通过View方式来实现逻辑层面的DWD，通过这种方式为下游Ad-hoc（数据探索）提供宽表查询支持，Doris的谓词下推及View优化的能力，为使用视图查询带来了便利。而当业务较重时，通过实体表单+微批任务进行实现，按照调度依赖关系逐层完成计算，针对使用场景对表单进行优化。

3）数据计算时效：新架构下的数据时效受具体数据计算链路中三个方面限制，分别是数据采集时效、批次计算时效、数据查询耗时。在不考虑网络吞吐、消息积压、资源抢占的情况下：

（实施类项目经常会遇到第三方不提供 Binlog 的情况，所以这里把通过批次采集数据也作为一个 case 列出来）

在Doris中为了达到更好的计算时效，基于Doris的数据计算流程相比在Hive中的计算流程可以进行一定的简化，这样可避免过多的冗余计算设计，以此提高计算产出效率。

4）补充架构能力：

Hadoop：根据不同的项目资源及数据情况来决定是否引入Hadoop 补充大规模离线计算场景。以实施类项目为例，Doris可以涵盖大部分核心业务数据计算场景。

MySQL：基于预计算的结果数据可以推送到下游MySQL 中以供 Report 查询，从而分散 Doris 计算查询的资源消耗，这样可以将资源充分留给核心且时效性要求高的应用或高频批次任务。如果计算资源充足，Doris也可以直接作为应用层的加速查询DB，而无须引入其它 DB。

4.2 新架构收益

通过引入Doris，我们成功构建了高时效、低成本的数据中台2.0，并成功满足了交付型项目和 SaaS 产品两种需求场景下的使用需求。新架构的收益如下：

数据时效性提升：架构1.0中大部分业务为 T+1 的支持方式，而在新架构下大部分业务都可实现实时或小时级的计算支持
资源利用率提高：在架构1.0中，离线资源在白天大部分时间处于闲置状态。而在新架构下，数据同步，计算（增量/全量）和查询均在同一集群下完成，从而提高了资源利用率。相较于部署一套 CDH，同等资源成本下，部署一套Doris可以带来更多的收益。
运维管理成本降低：在原有架构下，实时统计需求需要维护非常长的计算链路。而在新架构下，所有计算仅需在一个数据库中完成，更加简单、高效且易于维护。
易于业务扩展：Doris的节点扩展操作非常便捷，这对于业务的增量支持非常友好。

五、新架构的落地实践

我们在2022年底首次在测试环境中部署了Doris 1.1.5 版本，并进行了一些业务数据的导入测试和新架构的可行性验证。在测试后，我们决定在生产环境中落地实践 Doris。目前，新项目已升级到 1.2.4 版本并使用。Apache Doris 作为新架构下的核心系统，在整个架构中发挥着重要的作用。下面从模型选择、资源规划、表架构同步、计算场景实现、运维保障等几个角度分享基于 Doris 的项目落地经验。

5.1 模型选择

5.1.1 Unique模型

对于ODS层的表单来说，我们需要Doris Table与源系统数据保持实时同步。为了保证数据同步的一致性，我们采用了Unique 模型，该模型会根据主键来对数据进行合并。在1.2.0 版本推出之后，采用了新的 Merge On Write的数据更新方式，在 Unique Key 写入过程中，Doris会对新写入的数据和存量数据进行Merge操作，从而大幅优化查询性能。因此在使用 1.2 版本中，建议打开 Doris BE 的 Page Cache（在be.conf文件中增加配置项disable_storage_page_cache = false）。另外在很多情况洗，Unique 模型支持多种谓词的下推，这样表单也可以支持从源表直接建立视图的查询方式。

5.1.2 Aggregate模型

在某些场景下（如维度列和指标列固定的报表查询），用户只关心最终按维度聚合后的结果，而不需要明细数据的信息。针对这种情况，建议使用Aggregate 模型来创建表，该模型以维度列作为 Aggregate Key 建表。在导入数据时，Key列相同的行会聚合成一行（目前Doris支持 SUM、REPLACE、MIN、MAX 四种聚合方式）

Doris 会在三个阶段对数据进行聚合：

数据导入的ETL阶段，在每一批次导入的数据内部进行聚合；
底层BE进行数据Compaction合并阶段；
数据查询阶段

聚合完成之后，Doris 最终只会存储聚合后的数据，这种明细表单数据的预聚合处理大大减少了需要存储和管理的数据量。当新的明细数据导入时，它们会和表单中存储的聚合后的数据再进行聚合，以提供实时更新的聚合结果供用户查询。

5.2 资源管理

在生产环境中，我们使用一套 Doris 数据仓库支撑了多个下游数据应用系统的使用。这些应用系统对数据访问的资源消耗能力不同，对应的业务重要等级也不相同。为了能够更好管理应用资源的使用，避免资源冲突，我们需要对应用账号进行划分和资源规划，以保证多用户在同一 Doris 集群内进行数据操作时减少相互干扰。而Doris的多租户和资源隔离功能，可以帮助我们更合理地分配集群资源。Doris 对于资源隔离控制有两种方式，一是集群内节点级别的资源组划分，二是针对单个查询的资源限制。这里主要介绍下方式一的集群内节点级别的资源组划分过程。

1）第一步：需要梳理规划各场景的用途、重要等级及资源需求等，举例说明：

2）第二步：对节点资源进行划分、给节点打上 tag 标签：

alter system modify backend "10.10.101.1:9050" set ("tag.location" = "group_a");
alter system modify backend "10.10.101.2:9050" set ("tag.location" = "group_a");
alter system modify backend "10.10.101.3:9050" set ("tag.location" = "group_b");
alter system modify backend "10.10.101.4:9050" set ("tag.location" = "group_b");
alter system modify backend "10.10.101.5:9050" set ("tag.location" = "group_c");
alter system modify backend "10.10.101.6:9050" set ("tag.location" = "group_c");

3）第三步：给应用下的表单指定资源组分布，将用户数据的不同副本分布在不同资源组内

create table flume_etl<table>
(k1 int, k2 int)
distributed by hash(k1) buckets 1
properties(
    "replication_allocation"="tag.location.group_a:2, tag.location.group_b:1"
)

create table cdc_etl<table>
```   
    "replication_allocation"="tag.location.group_b:2, tag.location.group_c:1"

create table etl<table>
```
    "replication_allocation"="tag.location.group_a:1, tag.location.group_c:2"

create table mkui_readonly<table>
```
    "replication_allocation"="tag.location.group_a:2, tag.location.group_c:1"

create table SaaS_readonly<table>
```
    "replication_allocation"="tag.location.group_a:1, tag.location.group_b:1, tag.location.group_c:1"

create table dev<table>
`` `
    "replication_allocation"="tag.location.group_a:1, tag.location.group_b:1, tag.location.group_c:1"

4）第四步：设置用户的资源使用权限，来限制某一用户的查询只能使用其指定资源组中的节点来执行。

set property for 'flume_etl' 'resource_tags.location' = 'group_a';
set property for 'cdc_etl' 'resource_tags.location' = 'group_b';
set property for 'etl' 'resource_tags.location' = 'group_c';
set property for 'mkui_readonly' 'resource_tags.location' = 'group_a';
set property for 'SaaS_readonly' 'resource_tags.location' = 'group_a, group_b, group_c';
set property for 'dev' 'resource_tags.location' = 'group_b';

值得一提的是，与社区交流中得知在即将发布的 Apache Doris 2.0 版本中还基于Pipeline执行引擎增加了 Workload Group 能力。该能力通过对 Workload 进行分组管理，以保证内存和 CPU 资源的精细化管控。

通过将 Query 与 Workload Group 相关联，可以限制单个 Query （查询）在BE节点上的CPU 和内存资源的百分比，并可以配置开启资源组的内存软限制。当集群资源紧张时，将自动 Kill 组内占用内存最大的若干个查询任务以减缓集群压力。当集群资源空闲时，一旦 Workload Group 使用资源超过预设值时，多个 Workload 将共享集群可用空闲资源并自动突破阙值，继续使用系统内存以保证查询任务的稳定执行。更详细的 Workload Group 介绍可以参考：