分享嘉宾｜王春波 高级数仓工程师 《Doris实时数仓实战》作者文字校对｜志明与数据出品社区｜DataFun
01
背景介绍
电商是移动互联网时代最重要的业务形式之一，目前主流的业务形态是B2C。在这个群雄逐鹿的年代，除了淘宝、京东、拼多多等头部电商以外，还活跃着众多的中小规模电商平台。笔者所在公司的电商APP就是其中一个，目前注册用户超过2亿，月活跃用户接近2000万。
电商平台以APP作为载体，最重要的数据就是以订单为核心的结构化数据和以日志流为核心的半结构化数据，这也互联网业务最典型的应用场景。
订单业务包括下单、支付、发货、物流、评价、退货等业务流程，但是都可以通过order_id串联起来，数据保存在关系型数据库中。我们这边通过MySQL分库分表方案承载订单相关的业务数据，目前积累了自系统上线以来的1.5亿订单，目前日增长订单数为10万左右。
点击流数据则是APP上所有用户的操作行为埋点记录数据，是源源不断产生的半结构化数据。由于前期对APP埋点和日志流数据做过治理，所以目前数据格式比较规范，数据输出也比较稳定。点击流数据包括30+固定字段和一个扩展json字段组成，固定字段包括设备信息、会话信息、用户信息、网络信息、埋点编码等，扩展json字段内容则根据实际的页面情况生成，不同的页面或者埋点对应的扩展信息不同。点击流数据每日增量在10亿左右，ORC格式占用存储在1.16T左右。
笔者接手电商数仓项目时，恰逢公司推进数据治理项目，准备重建电商数仓。在我接手之前，公司数仓按照不同的业务模块划分不同的数据集市，电商业务有专门的电商集市，但是内部数据加工逻辑比较复杂、没有明确的数据分层和清晰的数据处理逻辑，基本上是面向需求开发，重复逻辑比较多，数据一致性差。我接手电商数仓以后，按照标准的数仓分层重构了电商数仓，同步产出实时数据，满足了实时数据看板、自助分析数据集、双十一大屏、每日业绩播报等多个数据应用。恰逢最近经历了新一轮618大促的考验，因此予以总结，形成经验分享给其他数仓开发的同行。
02
技术选型
数据中台作为公司统一的数据平台，承载着全公司大数据集群平台的基础能力，包括离线集群Hive、Hdfs、Yarn、Spark和Presto，实时集群Flink、ClickHouse，以及相关工具如自助分析工具QuickBI、调度系统、画像系统、监控告警系统、基于Zepplin的统一数据查询平台等。
公司的离线大数据有400多台服务器，基于Yarn框架进行统一的资源管理，计算资源分为离线计算、实时计算、实时查询等不同的资源队列，其中离线计算目前以Spark为主，部分高优先级的任务或者时效性较高的任务已经切换到内部改造过的Presto计算引擎。目前公司大数据平台上运行的离线数据处理任务主要分为MySQL2Hive数据采集、Hive2Hive数据加工、Hive2MySQL数据分发和Hive2CK数据分发四种类型，任务数分别是1.0W、1.2W、6K、500。
实时数据处理分为实时数据采集、实时数据计算和实时数据查询三个方面。实时数据采集通过自动化配置，直接写入Hive数仓的rt_ods库，目前有接近1000张表；实时数据计算目前主要是交给Flink完成，目前线上运行的大约500个任务；实时数据查询包括MySQL和Clickhouse，接入数据量不确定。
早期的数据结果查询都是基于MySQL分库分表来实现，2021年底开始引入ClickHouse作为交互式查询引擎。选择ClickHouse的原因主要是由于查询性能快、查询稳定，只要设置合理的分区，单表数据量可以达到百亿甚至千亿级别。目前公司在多个业务线引入了ClickHouse集群，在大数据线，ClickHouse集群主要替代MySQL分库分表方案，来实现数据的快速实时查询。大数据线的ClickHouse集群由28台节点组成14主*2副本集群，每台节点84核CPU+256G内存。
03
电商离线数仓
离线数仓总体上分为三层，即ODS、DW和DM层。
ODS层也叫数据采集层，数据来源于源系统，保留源系统概貌，为上游逻辑层提供原始数据，隔离对源系统的影响。我们这边分为SNAP、ODS、History三个数据库，分别存放快照数据、增量追加数据和全量历史快照数据。对于全量采集的数据，直接抽取到SNAP库；对于增量采集的数据默认会按照修改日期抽取最近一天新增或者修改的数据，按日分区存入ODS库，然后按照库表主键合并去重写入SNAP库；对于有保存历史快照数据需求的表，我们还会将历史快照复制一份按日保存到History库。
DW层也叫数据仓库层，我们分为DIM、DWD和DWS三个库。
DIM库用于保存公共维度数据，例如商品、商户、供应商、用户基础信息等。
DWD层也叫明细模型层，数据来源于ODS层，根据上游提供原始数据，划分数据主题，对ODS层数据进行关联整合。DWD层用于保存业务明细数据，只做简单的数据加工和多表关联，得到按照主题域和数据域划分的明细数据表。
DWS层也叫轻度汇总层，数据主要来源于DWD层，以指标加工为核心，按照维度建模的思路，加工一致性指标和一致性维度。DWS层也包括宽表层，所以DWS通常可以划分为两步进行数据加工，第一步聚焦于指标计算，统一加工业务指标，第二步关联维度信息，形成大宽表。有时候会把大宽表叫做DWT层，但是我们这边没有严格的区分。DWS层的宽表通常都是同步到ClickHouse，用于自助分析或者固定报表查询。
DM层也叫集市层或者数据应用层，数据来源主要来自DWS层，可按业务和应用主题分类，满足特定应用查询。DM数据主要保存在Hive数仓的DM库和对接数据应用的MySQL库、ClickHouse库。对于数据量超过千万的明细数据分析，数据会直接同步到ClickHouse库；对于百万级以下的数据，则直接保存到MySQL数据库。此外，还有应用层的用户画像数据保存在HBase。DM层的大部分数据直接来源于DWS，也有有些数据是在DWS层的基础上进行二次加工，包括简单汇总、计算同环比、多维汇总等，先写入DM层，再同步到外部数据库。
具体到电商数仓模块，我们主要构建了以下几个模型表：
当然，实际项目上设计和建设的模型远不止这几张表，我们还针对售后订单创建单独的表、根据埋点业务的运营位曝光和点击计算下单成交率、根据商品的推荐计算推荐模型的有效性，根据搜索的结果及点击计算不同入口的搜索成交情况等等。但是项目主要的核心的订单和点击数据流就是这10张表，其中商品标签表和用户标签表还作为电商业务商品画像和用户画像的基础数据来源表，提供画像标签的统一出口。
04
电商实时数仓
在离线数据加工的基础上，业务用户提出来实时数据的需求，主要包括企业微信业绩播报机器人和实时交易看板、实时成交监控、双十一大屏等。
最开始开发的是企业微信业绩播报机器人需求，每小时汇总一次当日成交数据，并和历史成交进行对比，将数据写入MySQL，再由Java程序读取数据，按照指定的数据格式播报到企业微信群。
针对这个业务场景，我们按照典型的Lambda架构设计，复用公司的Kafka写入Hive数据组件，通过配置化实现关键业务数据自动同步到Hive的rt_ods数据库。然后我们通过Presto计算引擎简化订单业务的加工逻辑，只计算关键成交指标，加工到DWS，并和离线数据加工的DWS层数据进行合并去重，保留最近13个月的订单明细。点击流数据不需去重，只保留当日、上月环期和去年同期三个日期的明细数据，并加工好关键指标，保留明细数据。最后一步是加工计算本期、同期、环期的不同指标，并分别按照商品维度和用户维度进行数据汇总，写入MySQL供JAVA应用查询。
第一代实时数仓架构
将实时播报任务串联成工作流，按照一小时一次的频率执行，截图如下：
实时播报满足了业务用户跟踪业绩进展的需求，但是时效性比较差，无法满足实时成交监控、实时看板和大促大屏的需求，于是我们又进一步开发了新的实时链路，即Flink实时链路。
第二代实时数仓架构
Flink实时链路主要由两个FlinkSQL任务组成，分别读取订单CDC日志流数据和点击埋点日志流数据，在进行简单的数据转换以后关联离线加工的商品信息表（定时同步到HBase，全量1600万）获取商品维度然后写入Clickhouse。在电商业务的多维分析中，最主要的维度就是商品维度和用户维度，其中商品维度包括商户信息、商品层级信息、商品规格信息、商品业务归属、商品价格和进货渠道等，用户维度包括用户注册信息、用户基本属性、用户成交记录和用户衍生标签。在我们的业务场景中，商品维度是千万级别，用户维度是亿级别，经过测试，在实时点击流中，由于数据流量比较大，关联用户信息会出现查询超时导致关联不上的场景，因为我们砍掉了实时数据的用户维度，而选择在ClickHouse进行结果数据查询时再利用Local Join的优势来关联用户维度。实时加工数据在ClickHouse中设置的TTL时间是3天，即仅保留最近三天的实时数据。
Flink实时链路的关键在于ClickHouse,我们首先将离线加工好的订单宽表、点击流宽表和用户维度信息表在每天跑批完成以后同步到Clickhouse(其中订单宽表是每日全量同步最近三个自然年的数据，点击流每日增量同步昨日数据)，然后通过一个视图来合并离线数据和实时数据，对外提供纯实时的一致性数据结果。
在ClickHouse这边主要处理逻辑有以下几点：
1.离线数据取下单（点击）日期小于当日的数据，实时数据取离线数据没有的下单（点击）日期对应的数据。这是为了避免在凌晨时离线数据还没有跑出来，导致查询昨日没有数据的情况。
2.实时数据关联用户维度，取用户注册时间和用户引流渠道等信息。基于ClickHouse的特性，我们将所有接入的数据默认按照fuid的hash值进行数据分片，确保同一个用户的订单、点击数据和用户维度数据在同一个数据分片上，既可以实现Join的本地化，又能减少用户数去重计算的资源消耗。为了强制join在本地进行，我们会直接在SQL中使用右表的local表进行关联。
3.根据订单和点击流的不同特点，承接订单实时数据的表我们采用ReplicatedReplacingMergeTree引擎表，点击流实时数据表则采用ReplicatedMergeTree引擎表。在使用订单实时数据时，我们会在表名后增加final关键字，确保读取到最新的数据。
Flink实时数据由于实时性高、数据完整度高并且基本上都是明细数据，可以满足各种业务场景，因此在这个数据集基础上我们创建实时成交看板、实时监控预警和大促大屏等应用需求。下一部分，我们将具体展开数据应用场景的方案解读。
05
数据应用
在电商数仓的基础上，我们构建了自助分析、固定报表、企业微信播报、标签画像、大促大屏等多个数据应用。其中，自助分析和固定报表都是基于QuickBI实现的，企业微信播报是Java程序，标签画像是自研系统，大促大屏是基于VUE开发的Web应用。
首先是自助分析，我们基于订单数据和点击流数据各自构建了一个宽表并同步到ClickHouse，不同的类目运营用户和数据产品都可以基于这两个自主数据构建自己的看板，并分享给其他同事。自助分析数据集根据用户的需求还在不停的追加字段，完成各种实验场景分析、用户成交分析和经营利润分析。订单宽表已经扩充到了256个字段，还有不少的用户标签和商品标签封装在fuser_label_json和fsku_label_json两个json字段中。目前，订单自助数据集是使用用户最多，应用最广泛的数据集。
其次是固定报表。在自助分析数据集的基础上，我们构建了业务经营日报、KPI进度监控等固定报表，满足管理层经营数据分析需求。这些报表主要在同环比、日周月年等维度上有一些特殊处理，导致需要做一些定制化开发，所以由我们数仓完成。
第三个应用是前面提到的企业微信播报，这里只截取其中一部分内容展现。企业微信从早上9点到晚上24点，每小时播报一次。其中最难的是24点以后的那次播报，需要做很多特殊处理，才能实现。
第四个应用是标签画像。我们的标签画像系统支持用户和商品两个维度，在标签系统定义的基础标签都会换成成SparkSQL，加工以后同步到HBase。衍生标签在基础标签的基础上组合定义，结果数据也会加工到HBase。
标签系统提供单个用户查询标签值和标签组合圈选用户两个功能，前者用于在线接口调用，后者用于导出用户进行分析或者广告投放。
第五个应用是大促大屏。我们参照阿里双十一大屏，构建了实时大促大屏，包括实时成交额、大促期间累计成交额、用户分类成交金额及本同期对比、商品分类成交金额及本同期对比。
06
后续演进和流批一体探索
目前第二代实时架构已经稳定运行了接近一年时间，做过一些修修补补的微调，但是整体架构没有变动过。这中间遇到的痛点主要有：
①离线数据跑完以后，昨日的实时成交数据会提高，但是第三天又会下降。这是因为离线数据是以12点作为快照时间点计算的，后面的成交或者退款数据在实时里面可以体现，但是离线需要到第三天才能体现。这个问题在大促期间暴露比较明显。
②商品维度数据一天只更新一次，导致当日上线的商品在统计时丢失，或者商品层级调整不能实时体现到看板中。
③流处理SQL封装在Flink管理平台中，批处理SQL封装在调度平台，导致两边容易出现逻辑不一致的情况。
④点击流数据积累过多以后，ClickHouse存储和查询性能出现瓶颈，但是集群扩容又比较困难，导致我们点击流数据只保留最近半年数据并且一次最多查询一个月的数据，用户满意度降低。
⑤维度变更导致点击流数据统计出现异常，比如商品类目、用户分类等。
面对以上这些问题，我们开启了第三代实时架构的设计和验证之路。第三代实时架构我们引入了基于数据湖的流批一体模式和基于OLAP数据库的多维实时数据查询模式。在数据湖方面，经过多方对比，最终选择Hudi作为数据湖底座，继续沿用Flink进行流式数据加工，选择Doris作为查询引擎。
选择Hudi的原因是：
①数据湖技术中Hudi目前最成熟，并且有很多案例分享；
②Hudi支持流式数据写入和流式数据读取，可以满足我们保存中间过程数据的需求；
③Hudi支持索引，可以更快的检索数据。
④Hudi和当前的HDFS存储底座结合更好。
选择Doris的原因有很多，例如支持多表关联、方便扩展、支持多种数据模型、支持多种索引机制和查询优化，还支持存算分离迁移历史数据到对象存储，直接查询外部数据源。更详细的关于Doris的特点和使用方法，欢迎购买笔者撰写的《Doris实时数仓实战》一书。有了Doris，最大的好处是我们可以做到维度解耦，可以在查询的时候才进行关联，一方面减少了数据存储空间占用，另一方面避免了历史维度不一致的情况。
总体来说，我们的第三代实时数仓架构如下：
采用这种流批一体的架构，可以解决流数据和实时数据切换时成交金额回退的情况；同时保留中间过程数据，可以逻辑变更导致需要回溯历史重算的情况；引入独立的OLAP查询引擎，可以解决查询的性能问题和多表关联问题。
虽然理想状态下，所有数据都通过Flink流式进行加工，但是经过调研，我们还是有一些数据的逻辑无法做到纯流处理，比如用户下单时是新用户还是老用户，所以我们还是保留了批处理的链路，批处理的数据通过Spark加工完成以后，直接写入Doris更新主键模型的部分列。
目前这个方案已经验证完成，正在进行配套平台的搭建和持续运行监控，预计Q4会全面铺开应用。
相比第一代架构和第二代架构，我们最大的特点就是用Doris代替了ClickHouse。虽然ClickHouse快并且稳定，但是其使用门槛较高、扩展性较差，元数据用ZooKeeper管理，在我们这边已经达到瓶颈，并且在实时数据高频写入的场景容易出现元数据管理异常导致数据写入失败的情况。
Apache Doris作为一款国产开源数据库软件，不仅实现了向量化引擎、存算分离、Merge on Write等前沿功能，还开创性的融合了数据分桶、行列混合存储、多种索引支持、多种数据模型等功能到MPP数据库中，是OLAP和数据仓库领域冉冉升起的新星。《Doris实时数仓实战》一书囊括Doris的基本操作、架构设计、进阶使用、运维管理、拓展应用等各方面的内容，还有大量的具体项目实践经验分析，非常适合想使用Doris进行数仓开发的小伙伴学习。目前Doris社区非常活跃，功能还在不断迭代和演化中，Doris背后的商业公司飞轮科技也给予了Doris发展非常大的助力，Apache Doris一定能成为一款具有全球影响力的开源产品。
今天的分享就到这里，谢谢大家。
网友评论