流批一体的近实时数仓的思考与设计

流批一体的近实时数仓的思考与设计导读：本文基于对数据时间旅行的思考，引出了对目前三种数仓形态和两种数仓架构的思考。结合数据湖在Flink的应用和数据湖元数据类型的思考，探索了基于数据湖的FlinkSQL流批一体的实践，在流批一体SQL表达一致、结果一致性、流批任务分离、混合调度依赖等进行了设计和探索。01数据的时间旅行和业务对数据的本质要求大规模的数据处理兴起于Hadoop生态的发展，关键在于分布式储存和分布式计算的发展，造就了如今近百种有关大数据的生态技术。数仓理论和建模理论基于大数据技术体系得以快速发展，其中离线数仓的标准化建设得到了广泛应用。数据的本质是一种行为的具象，业务在对数据的需求，核心在于对行为的可探索和可观察。基于此，我们需要明确一点，大数据技术是否完全满足了业务对数据需求在时间维度上的确定性了呢，这点是值得思考的。那么我们先来看一下数据的时间旅行。业务期望的数据：用户空间下的时间数据，t1时间数据，用户自然时间点或自然时间段的明细或者统计数据。

传输延迟：App用户，数据发送到网关或者日志服务系统，或者Server日志落文件系统所产生的延迟。Event进入到存储空间，可以代表数据已经是确定的，基本可观察，一般情况下，这个延迟很小。但是，在某些情况，比如APP的日志产生之后，但是因为网络等问题一直没有发送，或者Server宕机，导致延迟发送或者最终丢失。总体而言，传输延迟属于不可控延迟，暂时没有什么好的技术方案来解决。

存储空间：数据承载于实际的存储中，离线数仓承载于具体的分布式文件系统，实时数仓基于Kafka的消息队列系统，近实时数仓承载于数据湖存储中。这里可以抽象来看离线数仓，Event承载于分布式文件系统，以小时分区为例，某个小时的分区本质是自然时间产生的文件的集合，时间精度退化为小时级别。

计算延迟：数据进入存储之后，与进入计算空间的时间差，t3-t2。实时数仓中，计算延迟是数据的ProcessTime-IngestTime。离线数仓中，计算延迟是调度产生实例运行时间-数据进入存储空间的时间差。本质离线数仓和实时数仓的计算延迟在抽象上看是一致的。计算延迟在不同的数仓体系下，产生的时效不同，我们会划分为三种主流的数仓体系，秒级的实时数仓，分钟级的近实时数仓，小时级的离线数仓。可以看出，数仓的时效性差异，因为传输延迟的不可控，退化为计算延迟的差异。02离线、近实时、实时三种数仓在时间维度下的成因

在离线数仓和实时数仓，常常会提到数据的有界和无界，认为离线数仓的数据是有界的，实时数仓的消息流是无界的。准确与否在于数据的确定性考量。

离线数仓的确定性，在于文件自然生成时间的确定性和不可更改性，某个小时的自然文件生成，近似等于事件时间在自然时间的确定性，反例就是我们能看到数据漂移的情况，事件时间会或多或少落入上个小时或者下个小时的自然文件生成时间。那么离线数仓的确定性，实质是数据的IngestTime的确定性，具有天然的文件属性，易于分割。当我们说离线数仓计算的数据是准确的时候，默认了传输延迟带来的影响很小或者默认了当前小时的数据指标的标准是文件的自然形成时间。实时数仓，常常会提及不确定性或者说Lambda架构实际是对实时数仓的不确定性的替代方案。这种不确定性的原因是什么呢？这里分为四类情况说明，一是ETL的处理，从窗口上来说，是单条数据即为一个窗口，窗口的产生和销毁在一个Event中完成，y=window(data)。二是基于EventTime的时间窗口，如果再定义延迟时间，y=window(datas,datas.EventTime,delay)，第三种和第四种分别就是IngestTime和ProcessTime的时间窗口函数。对比离线数仓，可以看出，基于IngestTime的时间窗口和离线数仓的时间语义最为一致。离线数仓在时间窗口上，可以看做为数据进入文件的自然时间所对应的小时窗口，数据所承载的文件的确定性，保证了小时窗口的数据确定性，y=window(files)。近实时数仓，比如基于Iceberg的数据湖建立的近实时数仓，在于离线数仓对比中，实际是将基于小时文件细分到分钟级别的快照文件上来，y=window(snapshots)。对比实时数仓，因为Kafka的IngestTime目前在精确性上是不精确的，基于快照的文件划分，在精确性上有一定的保证，同时在降低时效程度，从秒退化为分钟，很多场景是可以容忍的。三种在时间维度对比上看，一是在某个时间，统计的本质对业务的需求都是近似的，这个本质是传输延迟所带来的，但是这个在实践中，不影响数据的可用性和统计学意义。二是不同数仓的划分，是存储和计算技术发展所带来的。三是离线数仓的确定性模糊了传输延迟，实时数仓的不确定性，是对传输延迟的一种取舍，人为的限定了EventTime的最大延迟时间，从而保证了指标的时效性，都是具有实践的意义所在。03Lambda和Kappa架构在时间维度下的取舍

当离线数仓刚刚发展的时候，只有一种数仓架构，也是基于大数据分布式处理刚刚发展的原因。随着实时技术的发展，大家在时效性上有了更多要求，但是同离线数仓对比的时候，在数据的准确性上，因为统计的窗口不同，必然会导致某个时刻的指标结果的不严格一致。

为了解决这种不严格一致的情况，Lambda架构（由Storm的作者NathanMarz提出的）产生的，实时确保时效，离线确保准确。最终会以确保离线三个时间窗口的统计一个事件时间窗口的结果，来回补实时数仓以为EventTime窗口，因为时效性丢弃的延迟数据的结果，从而保证业务上对EventTime窗口的要求，或者默认为离线的IngestTime所产生的文件分区近似认为EventTime的时间窗口。这种带来的弊端，维护两套数据路线，而大家总在想办法解决。

Kappa架构的提出，得益于实时计算的效率提升，但是因为在批处理技术短板，生产实践推广受限。Kappa架构是基于实时EventTime的一种数据窗口处理，因为Kafka的IngestTime不精确和为了同离线数仓对比而权衡考虑，EventTime在传输延迟上的不可控，导致Kappa架构的准确性就会出现折扣。虽然是业务上最准确的时间范围，可行性上确不佳。

近些年来，不断发展的MPP架构的OLAP查询引擎，并不会涉及到时间窗口的计算取舍，OLAP引擎本质是基于ProcessTime来加速查询的一种技术手段，是数仓不可分割的一部分，但是传输延迟的不可控没有解决，但是将计算延迟下推到了查询时，通过快速查询来解决尽可能减少计算延迟，同时保证了查询的灵活性，自助分析探索上有着广泛的应用。

从数仓架构的发展上看，不断在围绕结果的确定性，技术的可行性，数据的时效性，查询的灵活性上，不断的权衡，各个组件也是依据实际需求而发展起来的。04数仓一体的可行性思考

基于三种数仓体系和两种架构的思考，每个设计都是兼顾一种或多种考量，那么能不能实现一种机制，能够较好的满足数仓需求体系建设呢？从目前的技术发展上看，是有一定的可能性的。架构体系的发展一是基于技术基础，二是不断吸收组件的优点，做加法。

除去实时、近实时、离线数仓的划分，从技术的视角去看数仓建设的可行性。那么我们就要选取一些重要的点，取舍掉一些不可能的实现。

第一点是结果的确定性，这点是基于离线数仓发展的思考。不确定性带来的问题是信息的不对称，确定性的结果是可以模糊一定的指标含义的。

第二点是数据的时效性，高时效必然能够满足低时效，反之不然。另外数据的时效性，本身是基础组件的技术发展所限制的。

第三点是开发的便利性，排在时效性后面的考虑是，便利性是基于应用层面建设的，难度一般是弱于基础组件的，可以通过不断实践优化，达到一个良好的使用体验。第四点是查询的灵活性和高响应，OLAP的基础设计保证了查询速度，那么OLAP的技术架构体现是可以复用或者拓展的。

那么基于上面四点考虑，可以在实时数仓的基础上，优先解决掉确定性问题。这个是很重要的一个命题，要保证计算结果同离线数仓的一致性。这一点的实现方面，可以参考离线数仓，模糊EventTime和IngestTime，用文件的start和end作为确定性的依据，文件的中间实时计算，确保时效性。那么基于Flink，就需要实现一种基于文件自然分割的Watermark机制，作为计算窗口划分的依据。

在确定性问题之后，需要解决计算的成本和使用的成本，这里比较重要的是存储层，实时数仓依赖Kafka，Kafka发展不具备数仓一些重要的点，成本是一个方面，查询是一个方面，Kafka无法架构在各种OLAP引擎或者计算引擎上面。这里，近实时数仓的依赖，比如数据湖或者Paimon，数据湖分钟级的时效。不过，从发展的角度上看，是一种可行的解决方案。数据湖兼顾了流计算和批计算，同时，如果未来OLAP引擎如果能够在数据湖上实现类似MPP架构的查询效率，这也是有可能的，比如短期可以用数据冗余，将数据湖格式的数据转换一份到OLAP对应的引擎上实现加速查询。

第三个方面，流式计算的管理和依赖机制，借鉴于离线数仓的管理方式，需要一套完备的数据依赖管理，任务容错回跑机制。实时数仓一般是基于单个任务式的管理，离线数仓是基于任务流的管理，那么实时数仓的发展，也必然要实现任务流的管理方式，覆盖整个开发链路。

为了实现一种统计的数仓架构，那么需要的发展工作如下：一是着重发展存储层，比如数据湖，既要比较好的适应流和批引擎，又要能够高度适应OLAP查询引擎。二是在实时数仓或者近实时数仓，引入类似离线数仓的调度依赖管理和补数和容错回跑机制，或者在离线调度上兼容流任务依赖调度，实现任务流级别的管理和流批一体的数仓实现。三是在引擎层着重发展Flink批处理能力。最终的任务运行方式同时包含三种：实时模式、离线模式、业务模式，分别对应着不同的数据准确性级别。也可以任选其一或者其二作为运行方式。05基于Flink和数据湖的流批一体近实时数仓设计示例

数仓任务在离线调度和实时任务的简单抽象示例:数据源=>同步任务/实时任务=>stg_table(partition=hour)=>计算任务(insertoverwritepartition=hour)=>dwd_table(partition=hour)=>计算任务(insertoverwritepartition=hour)=>dws_table(partition=hour)=>同步任务=>OLAP加速=>数据服务

如果存储层是基于数据湖（以Paimon为例）：离线调度产生的表的版本信息，commit_kind:insertoverwrite类型的。同时离线任务的驱动，是基于调度依赖的驱动，onebyone的调度。

如果是基于流式计算，比如分钟级生成snapshot那么会演变为：数据源=>同步任务/实时任务=>stg_table(version=snapshot_id)=>计算任务(insertintoversion=snapshot_id)=>dwd_table(version=snapshot_id)=>计算任务(insertintoversion=snapshot_id)=>dws_table(version=snapshot_id)=>同步任务=>OLAP加速=>数据服务那么启动多个任务，任务是持续的运行。commit_kind:insertinto类型的。

那么要想实现流批一体的近实时数仓，需要解决如下问题：

1.Flink任务支持批量计算能力要持续不断的加强

从Flink1.16/1.17的版本发布情况，在批处理能力上有比较大的提升，同时，社区也在持续不断的加强批处理能力以及同hive的兼容能力。

2.如何使用同一份FlinkSQL，既可以用于批任务调度，又可以用于流任务运行呢两张表：dwd_partition_word_count，dws_partition_word_count，计算wordcountCREATETABLEtablestore.tablestore_test.dwd_partition_word_count(logdateString,user_idbigint)PARTITIONEDBY(logdate)WITH('bucket'='3');

CREATETABLEtablestore.tablestore_test.dws_partition_word_count(logdateString,user_idbigint,cntBIGINT,PRIMARYKEY(logdate,user_id)NOTENFORCED)PARTITIONEDBY(logdate)WITH('bucket'='3');批任务的FlinkSQL：insertoverwritetablestore.tablestore_test.dws_partition_word_countPARTITION(logdate=${start_date})selectuser_id,count(1)ascntfromtablestore.tablestore_test.dwd_partition_word_countwherelogdate=${start_date}groupbyuser_id;--或者insertoverwritetablestore.tablestore_test.dws_partition_word_countselectlogdate,user_id,count(1)ascntfromtablestore.tablestore_test.dwd_partition_word_countwherelogdate=${start_date}groupbylogdate,user_id;流任务的FlinkSQL:insertintotablestore.tablestore_test.dws_partition_word_countselectlogdate,user_id,count(1)ascntfromtablestore.tablestore_test.dwd_partition_word_countgroupbylogdate,user_id;

如何用一个FlinkSQL来实现流批模型下的不同呢？

不同点：Insertinto和Insertoverwrite的问题，这个通过在提交运行模式的时候，如果是批任务，则是InsertOverwrite，如果是流任务，则转为Insertinto，这个在技术上没有什么难点。

不同点：Where条件的数据范围问题。抽象来看，流任务和批任务的时间范围在表达上是可以统一的insertoverwritetablestore.tablestore_test.dws_partition_word_countselectlogdate,user_id,count(1)ascntfromtablestore.tablestore_test.dwd_partition_word_countwherelogdate>=${start_date}andlogdate<=${end_date}groupbylogdate,user_id;

比如跑4月22号一天的数据，执行的批SQL为：insertoverwritetablestore.tablestore_test.dws_partition_word_countselect

logdate,

user_id,count(1)

as

cnt

from

tablestore.tablestore_test.dwd_partition_word_count

where

logdate>='20230422'

and

logdate<='20230422'

group

by

logdate,user_id;

如果用流模式跑，执行的SQL可以为：select

logdate,

user_id,count(1)

as

cnt

from

tablestore.tablestore_test.dwd_partition_word_count

where

logdate>='19700101'

and

logdate<='99990101'

group

by

logdate,user_id;

insertoverwrite/into和时间范围，可以由平台执行的时候自动转换和参数输入。

3.批任务的调度和流任务的计算如何分离

任务完成开发，在批模式下，用调度任务验证了逻辑无误，那么之后可以用流模式，一直持续不断的运行。一是计算逻辑变更或者历史数据修复怎么办，二是可不可以支持流批双跑。其实本质是一个问题。如果计算逻辑变更，那么可以修改流批一体的SQL逻辑，然后流任务