三、解决方案

原文大佬的这篇DorisFlink构建实时数仓的实战文章整体写的很深入这里直接摘抄下来用作学习和知识沉淀。

本篇文章介绍如何基于Doris和Flink快速构建一个极速易用的实时数仓包括数据同步、数据集成、数仓分层、数据更新、性能提升等方面的具体应用方案。

先介绍一下传统的数据架构如何设计的、又存在哪些痛点问题。

下图为传统的数据架构如果从数据流的角度分析传统的数据处理架构会发现从源端采集到的业务数据和日志数据主要分为实时和离线两条链路

在实时数据部分通过Binlog方式将业务数据库中的数据变更CDCChange

Data

-Kafka-Sink对日志数据进行实时采集当不同来源的数据都采集到实时存储系统后便可以基于实时存储系统来构建实时数仓。

在实时数仓的内部仍然会遵守传统数仓分层理论将数据分为

ODS

层以实现最大程度的模型复用。

在离线数据部分通过DataX定时同步的方式批量同步业务库RDS中的数据。

当不同来源的数据进入到离线数仓后便可以在离线数仓内部依赖Spark

SQL

对象存储上。

通过这样的⽅式离线数仓便构建起来了。

与此同时为了保障数据的一致性通常需要开启数据清洗任务使用离线数据对实时数据进行清洗和定期覆盖保障数据最终的一致性。

从技术架构的角度对传统数据技术栈进行分析会发现为了迎合不同场景的需求往往会采用多种技术栈例如对于

OLAP

Druid。

除此之外为应对半结构化数据的分析需求例如日志分析与检索场景通常会使⽤

Data

HBase等。

其中涉及到的数据组件有数十种高昂的使用成本和组件间兼容、维护及扩展带来的繁重压力成为企业必须要面临的问题。

传统数据架构组件繁多维护复杂运维难度非常高。

计算、存储和研发成本都较高与行业降本提效的趋势背道而驰。

同时维护两套数据仓库实时数仓和离线数仓和两套计算实时数据量和实时计算任务数据时效性和一致性无法保证。

在此背景下需要构建⼀个“极速、易用、统一、实时”的数据架构来解决这些痛点

极速更快的查询速度最大化提升业务分析人员的效率易用对于用户侧的使用和运维侧的管控都提供了极简的使用体验统一异构数据与分析场景的统一半结构化和结构化数据可以统一存储多分析场景可以统一技术栈实时端到端的高时效性保证发挥实时数据的价值

采取Doris和Flink

来构建极速易用的实时数仓具体架构如下图所示。

多种数据源的数据经过flink

cdc

job加工件处理后入库到Doris或者Hive等湖仓中最终基于Doris提供统一的查询服务。

Load将kafka等消息系统的数据实时同步到Doris在数仓分层上ODS层通常选择明细模型构建DWD层可以通过SQL调度任务对ODS数据抽取并获取DWS和ADS层则可以通过物化视图和Rollup进行构建。

在数据湖上

Doris

Laketodo提供联邦分析和湖仓加速的能⼒。

在数据应用上Apache

Doris

BinLog的实时增量数据同步。

全量数据同步和增量数据同步可以自动切换因此在数据迁移过程中只需要配置好同步的表即可。

当Flink任务启动时优先进行历史表的数据同步同步完成后自动切换成实时同步。

3.1.2

如何保证数据一致性是大家重点关注的问题之一那么在新架构是如何实现的呢

数据⼀致性⼀般分为“最多⼀次”

最多⼀次At-Most-Once发送⽅仅发送消息不期待任何回复。

在这种模型中数据的⽣产和消费过程中可能出现数据丢失的问题。

⾄少⼀次At-Least-Once发送⽅不断重试直到对⽅收到为⽌。

在这个模型中⽣产和消费过程都可能出现数据重复。

精确⼀次Exactly-Once能够保证消息只被严格发送⼀次并且只被严格处理⼀次。

这种数据模型能够严格保证数据⽣产和消费过程中的准确⼀致性。

Flink

Job启动及Doris事务开启当Flink任务启动后Doris

Sink

Doris。

事务预提交当Flink开始进行Checkpoint时Flink会发起Checkpoint请求此时Flink各个算子会进行Barrier对齐和快照保存Doris

Sink发出停止

写⼊的请求并发起一个事务提交请求到Doris。

这步完成后这批数据已经完全写入Doris

BE中但是BE没有进行数据发布前对用户是不可见的。

事务提交当Flink的Checkpoint完成之后将通知各个算子Doris发起一次事务提交到Doris

Flink

两阶段事务提交保证了数据写入一致性。

需要注意的是在该过程中可能遇到一个问题如果事务预提交成功、但

Flink

内部支持对写⼊数据进⾏回滚(Rollback)从⽽保证数据最终的⼀致性。

3.1.3

Schema的变更需求。

当RDS表结构发生变更时用户期望Flink

RDS

Change的实现原理相对简单对数据表的加减列操作不再需要同步更改数据文件仅需要再FE中更新元数据即可从而实现毫秒级的Schema

Change

解决了数据同步过程中源数据⼀致性的保证、全量数据和增量数据的同步以及

DDL

将Mysql中两个表的数据同步到Flink后在Flink内部进行多流Join完成数据打宽后将大宽表同步到Doris中。

对上游的Kafka数据进行清洗在Flinkjob完成清洗后通过Doris-Sink写入到Doris中。

对Mysql数据和Kafka数据在Flink内部进行多流Join将Join后的宽表结果写入Doris中。

在Doris侧预先创建宽表将上游RDS中的数据根据Key写入使用Doris的部分列更新将多列数据写入到Doris的大宽表中。

3.3

Doris针对不同场景提供了不同的数据模型分别为明细模型、聚合模型、主键模型。

3.3.1

在某些多维分析场景下数据既没有主键也没有聚合需求Duplicate

数据模型可以满足这类需求。

明细模型主要用于需要保留原始数据的场景如日志分析用户行为分析等场景。

明细模型适合任意维度的

Ad-hoc

查询即席查询。

虽然同样无法利用预聚合的特性但是不受聚合模型的约束可以发挥列存模型的优势只读取相关列而不需要读取所有

Key

在企业实际业务中有很多需要对数据进行统计和汇总操作的场景如需要分析网站和

APP

访问流量、统计用户的访问总时长、访问总次数或者像厂商需要为广告主提供广告点击的总流量、展示总量、消费统计等指标。

在这些不需要召回明细数据的场景通常可以使用聚合模型比如上图中需要根据门店

3.3.3

在某些场景下用户对数据更新和数据全局唯一性有去重的需求通常使用UNIQUE

KEY

操作比如图中我们以订单id为唯一主键对订单上的其他数据时间和状态进行更新。

3.4

由于数据量级普遍较大如果直接查询数仓中的原始数据需要访问的表数量和底层文件的数量都较多不同层级对数据或指标做不同粒度的抽象通过复用数据模型来简化数据管理压力利用血缘关系来定位数据链路的异常同时进一步提升数据分析的效率在Doris

3.4.1

可以将原始表的数据进行处理和过滤并写入到目标表中这种SQL抽取数据的行为

一把是以微批形式进行例如15分钟一次的ETL计算任务通常发生在从ODS到dwd层数据的抽取过程中因此需要借助外部的调度工具例如Dolphinscheduler等来对ETL

3.4.2

物化视图本质是一个预先计算的过程可以在Base表上创建不同的物化视图或Rollup来对Base表进行聚合计算。

通常在明细层到汇总层例如dwd层到dws层或

dws层到ads层的汇聚过程中可以使用物化视图以此实现指标的高度聚合。

同时物化视图的计算是实时进行的因此站在计算的角度也可以将物化视图理解为一个单表上的实时计算过程。

3.4.3

的查询结果固化以供用户直接查询支持定时自动或手动触发的方式进行全量更新查询结果。

基于多表物化视图这一功能的实现可以做更复杂的数据流处理比如数据源侧有

TableA、TableB、TableC在多表物化视图的情况下用户就可以将

TableA

在实时数仓构建的过程中还需面临高并发写入和实时更新的挑战如何在亿级数据中快速找到需要更新的数据并对其进行更新⼀直都是大数据领域不断追寻的答案。

3.5.1

模型来满足数据更新的需求同时通过MVCC多版本并发机制来实现数据的读写隔离。

当新数据写入时如果不存在相同key的数据则会直接写入如果有相同key的数据则增加版本此时数据将多个版本的形式存在。

后台会启动异步的Compaction进程对历史版本的数据进行清理当用户在查询时Doris会将最新版本对应的数据返回给用户这种设计解决了海量数据的更新问题。

在Doris中提供了Merge-on-Read和Merge-on-Write两种数据更新模式。

Merge-on-Read

读取数据时发现最新版本的数据被标记删除就会将该数据从查询结果中进⾏过滤。

Merge-on-Read的特点是写入速度比较快但是在数据读取过程中由于需要进行多路归并排序存在着大量非必要的CPU计算资源消耗和IO开销。

在1.2.0

Merge-on-Write的数据更新模式。

Merge-on-Write兼顾了写入和查询性能在写入过程中引入了Delete

Bitmap数据结构使用Delete

Bitmap标记Rowset中某一行是否被删除为了保持Unique

Row

Bitmap将Bitmap中的MemTable一起存储在BE中每个Segment会对应⼀个

写入流程

Merge-on-Write该模式不需要在读取的时候通过归并排序来对主键进行去重这对于高频写入的场景而言大大减少了查询执行时的额外消耗。

此外还能够支持谓词下推并能够很好利用Doris丰富的索引在数据IO层面就能够进行充分的数据裁剪大大减少数据的读取量和计算量因此在很多场景的查询中都有非常明显的性能提升。

在真实场景的测试中通过

Merge-on-Write可以在保证数万QPS的高频Upsert

3-10

部分列更新是一个比较普遍的需求例如广告业务中需要在不同的时间点对同一个广告行为展示、点击、转换等数据的更新。

可以通过

Aggregate

物化视图除了可以作为高度聚合的汇总层外更广泛的定位是加速相对固定的聚合分析场景。

物化视图是指根据预定义的SQL分析语句执行预计算并将结算结果持久化到另一张对用户透明用户无感知但有实际存储的表中在需要同时查询聚合数据和明细数据以及匹配不同前缀索引的场景命中物化视图时可以获得更快的查询性能。

Base表可以构建多个不同的物化视图从不同的维度进⾏统计。

如果数据再物化视图中存在会直接查询物化视图如果在物化视图中不存在才会查询Base表。

在数据写入或更新时数据会在写入Base表的同时会写入物化视图从而保证物化视图和Base

智能路由选择遵循最小匹配原则只有查询的数据集⽐物化视图集合⼩时才可能⾛物化视图。

如上图所示智能选择过程包括选择最优和查询改写两个部分

选择最优

count(distinctuser_id)然后对数据进⾏计算由于

Count

由此过程可⻅如果通过物化视图进行查询查询效率更⾼。

当我们找到最优查询计划就可以进⾏⼦查询改写将

Count

分桶(Bucket)可以按照时间对数据进⾏分区再按照分桶列将⼀个分区的数据进行

Hash

分到不同的桶⾥。

在查询时则可以通过分区分桶裁剪来快速定位数据加速查询性能的同时实现高并发。

3.6.3

的MPP查询框架、向量化执行引擎以及查询优化器也提供了许多性能优化方式在此仅列出部分、不做详细展开

向量化引擎基于

内部构建不同的数据分层最后在上层构建不同的数据应⽤比如自助报表、数据⼤屏。

除此之外它基于应用平台构建了数据开发与治理平台完成了源数据管理、数据分析等操作。

使用收益

架构体系后架构简洁、组件减少解决了多架构下的数据的冗余存储服务器资源节省了

60%运营成本⼤幅降低。

基于该数仓架构在⽤户的业务场景上可以支持数万次的⽤户在线查询和分析。

4.3

等多个技术栈在该架构下查询性能无法得到有效快速的提升维护和开发成本一直居高不下。

使用收益

在Doris1.2版本中对数据表的加减列操作不需要同步更改数据文件仅需要在FE中更新元数据即可从而实现毫秒级的Schema

Change操作。

与此同时使得Doris在面对上游数据表维度变化时可以更加快速稳定实现表结构同步保证系统的高效且平稳运转。

通过

Flink