腾讯游戏广告流批一体实时湖仓建设实践

原创

可君

修改于 2023-01-10 20:33:36

1.2K0

修改于 2023-01-10 20:33:36

腾讯游戏广告业务对数据准确性和实时性均有诉求，因此数据开发团队分别搭建了离线及实时数仓。技术视角下，这是典型的Lambda架构，存在数据口径不一致、开发维护成本高等弊端。在降本增效的大背景下，我们针对结合计算引擎Flink与数据湖技术Iceberg建设流批一体实时湖仓做了较多的探索和实践，已经具备可落地可复制的经验。借助Flink框架支持批处理作业的能力，我们实现了将流处理层和批处理层的计算层面统一于Flink SQL，存储层面统一于Iceberg。

1. 离线及实时数仓建设历程

为了满足腾讯游戏广告业务对数据准确性及实时性的需求，我们先后搭建了离线数仓及实时数仓，分别输出数据延迟为日/小时级的指标与数据延迟为分钟级的指标，供用户在部门BI看板上查看。

1.1 离线数仓建设

首先我们使用Spark计算，Hive存储构建了离线数仓，这里和传统离线数仓区别不大，稍微特殊的一点是使用Spark将MongoDB、MySQL等异构数据源统一到内部的DataFrame中，然后写入Hive作为我们的维表层。技术视角的离线数仓架构如下：

1.2 实时数仓建设

接着我们使用Flink计算，Kafka、Redis、ClickHouse存储构建了实时数仓，在维表层数据同步上，我们使用了Flink CDC来实时监听数据源变化，实时同步到Redis中以快速地反映维度数据的变化。技术视角的实时数仓架构如下：

2. 数仓架构介绍及选型

谈到数仓架构，就离不开经典的Lambda和Kappa架构。关于这两者的具体优缺点，相关的文章已经很多了，下面从比较新颖的抽象模型角度对这两种架构进行介绍，并引申出各自的优缺点。最后谈谈我们如何基于这两者去做架构选型的。

2.1 Lambda架构

Storm 的作者南森·马茨（Nathan Marz）于2010年提出了 Lambda 架构，把大数据的批处理和流式处理结合在一起，变成一个统一的架构，它可以被称之为第一个“流批协同”的大数据处理架构。他把整个数据处理流程抽象成 View = Query(Data) 这样一个函数，其中原始日志就是我们的主数据（Master Data），不管是批处理还是流式处理，都是对于原始数据的一次查询（Query），得到的计算结果就是一个基于特定查询的视图（View）。在系统的整体架构上，只需要对外部的用户暴露出 View，屏蔽掉下面 Query 和 Master Data 的细节。在此基础上，Nathan总结的 Lambda 架构由以下三部分组成：

首先是输入数据，即 Master Data
然后是一个批处理层（Batch Layer）和一个实时处理层（Speed Layer），它们分别是一系列批处理任务，和一系列流式处理任务，获取相同的输入数据各自计算出结果
最后是一个服务层（Serving Layer），通常体现为一个数据库。批处理层和实时处理层的结果都会写入到这个数据库里。并且，后生成的批处理层的结果，会不断替代掉实时处理层的计算结果，也就是对于最终计算的数据进行修正

由上面的介绍可以看出，我们之前分别搭建的离线数仓和实时数仓，其实就是Lambda架构的设计。只是在最后的服务层中我们的具体实现并没有合并离线和实时的结果数据。

Lambda 架构很好地结合了批处理计算准确和流式处理计算延时低的优点，但是它也有一些缺点：

所有的视图，既需要在实时处理层计算一次，又要在批处理层计算一次。即使我们没有修改任何程序，也需要双倍的计算资源
我们所有的数据处理程序，也要撰写两遍。批处理层的程序和实时处理层的程序虽然要计算的是同样的视图，但是因为底层的框架完全不同，代码我们就需要写两套，也就是我们需要双倍的开发资源
因为批处理层和实时处理层的代码不同，两边对于同样视图的理解不同，采用了不同的数据处理逻辑，可能引起数据口径不一致

2.2 Kappa架构

Kafka 的作者杰伊·克雷普斯（Jay Kreps）于2014年提出了Kappa 架构，它可以被称之为第一代“流批一体”的大数据处理架构。Kappa 架构在 View = Query(Data) 这个基本的抽象理念上，和 Lambda 架构一致。但是相比于 Lambda 架构，Kappa 架构去掉了 Lambda 架构的批处理层，而是在实时处理层，支持了多个视图版本。在 Kappa 架构下，如果要对 Query 进行修改，可以按照以下步骤进行：

原来的实时处理层代码先不用动，另外部署一个新版本的代码
对这个新版本的代码进行对应日志的重放，在服务层生成视图的一个新的版本
在日志重放完成之前，外部的用户仍然查询旧的实时处理层产生的结果。一旦日志重放完成，新的代码处理到最新产生的日志，那么就可以让查询切换到新的视图版本上，旧的实时处理层的代码就可以停掉了

Kappa 架构放弃批处理层，转而在实时处理层提供多个程序版本的思路很好地规避了批处理层和实时处理层割裂的弊端，但是它也有一些缺点：

Kappa架构非常依赖于消息队列重放日志的能力，但是消息队列的存储存在瓶颈，对于需要回溯大量历史数据的场景无能为力，但是这类场景在日常需求中比较常见
消息队列中的中间结果数据很难使用常用的OLAP引擎进行高效查询，使得通过分析这类数据进行程序调试变得困难
流式处理中用于平衡流式数据乱序和数据实时性的Watermark机制以及多流join时不同流数据的乱序问题，很可能造成需要计算的数据丢失，从而影响计算结果的准确性

2.3 当我们谈流批一体时，我们在谈什么

Kappa架构希望只借助流式处理就同时满足之前对批处理和流式处理的需求，其实是将“批(Batch)”看成是“流(Streaming)”的一种特殊情况。反过来看，把批的记录数限制到了每批一条，那么它就是所谓的流了。从这个视角观察，流和批本身是同一个事物。进一步地，Google于2015年发表了《The Dataflow Model》的论文，对于流式数据处理模型做出了最好的总结和抽象，提出Dataflow模型。Dataflow将所有数据都视为“无边界（Unbounded）”的数据集，MapReduce的“有边界（Bounded）”的数据集，也只是 Dataflow 的一种特殊情况。值得注意的一点是，Dataflow提出的触发之后的输出策略中累积并撤回策略至今仍未被实现，因此流式计算结果的正确性，在有些情况下是保障不了的。从这个角度来看，Lambda 架构仍未彻底过时。

综上所述，Lambda架构和Kappa架构各自都有一些比较显著的缺陷，所以我们综合参考了两种架构来实现我们的流批一体数仓。主要思路是：总体架构沿用Lambda架构，分别通过批处理层和流处理层满足业务对于数据准确性及实时性的诉求。在此基础上希望借助Kappa架构看待数据流批一体的视角去改进Lambda架构，寻找一个实现了Dataflow模型的计算引擎去统一处理批处理层和流处理层的数据计算。与此同时，也希望寻找一个同时支持批处理全量读写和流处理增量读写的数据存储技术以统一批处理层和流处理层的数据存储，从框架层面保证数据的一致性以及节省各项成本。

3. 流批一体实时湖仓建设实践

在具体展开之前，从结果导向出发，先明确下我们期望流批一体最后实现的效果是什么。

从大的方面来说，大数据技术要回答的两个问题是：

（1）海量数据如何存储？

（2）海量数据如何计算？

具体到流批一体，这里可以细分为存储和计算两个层面，我们可以按照以下步骤去确定目标：

（1）存储层面流批一体，即通过一种统一的存储技术能在同一张表上同时支持流处理和批处理，以此达到“Single Source Of Truth”的目的。这样底层明细数据是同一份（具体对应数仓中的DWD层），数据天然具备一致性，同时避免了在批处理层和流处理层使用两套不同存储系统带来的存储成本增加

（2）计算层面流批一体，即我们写的同一套代码，只需要通过配置区分，并做相应细微调整就能同时用于流处理层和批处理层。这样从框架层面保证了计算逻辑一致性，而不是开发人员人工保证流处理层和批处理层不同计算框架的代码逻辑一致，同时节省了开发和维护成本。这里为什么要提到细微调整？其实是与我们的需求场景有关的。我们的批处理层需求和流处理层需求即使需要的维度和指标均一致，这两者计算的时间窗口也是不同的。反映到最终的结果表里就是两者标识“统计时间段”的字段不同，批处理层需求一般统计范围为天级或者小时级，流处理层需求则一般为分钟级或秒级（这里并不是说流处理不能计算天级需求或者批处理不能计算分钟级需求，只是在准确性或者实时性上无法满足）。对应到计算代码就是即使主要计算逻辑一致，分组字段中的“时间窗口”也是不同的，所以只能复用主要的计算逻辑，代码并不是完全相同

（3）存储和计算层面流批一体，兼具上述两者的优点

3.1 存储层面流批一体

存储层面流批一体需要有满足上述需求的存储技术支持，经过调研我们发现最近比较火热的数据湖技术Iceberg可以承担这个任务，并借在数仓中引入数据湖实现湖仓一体。需要注意的是，数据湖底层是基于HDFS等分布式存储之上的，用于构建流处理层时实时性不及消息队列，有分钟级延迟。不过我们的业务场景对实时数据延迟不敏感，分钟级延迟可以接受。这样，在存储统一到Iceberg的同时，仍然保留流处理层使用Flink计算，批处理层使用Spark计算，这是我们目前正在迁移改造中的存储层面流批一体的湖仓架构：

这里比较特别的一点是我们没有在DWD层就将流处理层和批处理层分开，而是使用Flink消费ODS层的消息队列数据，经过配置化的Flink DataStream API进行一系列统一模式的ETL处理后写入DWD层的Iceberg，然后从统一的DWD层数据出发分别构建了流处理层和批处理层的后续数仓分层。这里的主要考虑是我们的ODS层数据开始都是存放在Kafka中，且DWD层数据是一条一条的明细数据（未聚合）可以被流处理层和批处理层复用，没有必要分别使用Flink和Spark建设两遍DWD层。当然这里因为是Flink流处理任务可能存在运行不稳定问题，同时数仓上层建设完全依赖于DWD层明细数据，也就是DWD层是整个数仓的基座，必须保证数据最终是exactly once的。因此我们另外维护了Flink任务将ODS层Kafka数据写入Iceberg，同时利用Iceberg可以时间旅行及通过startSnapshotId和endSnapshotId增量读取，写入时可以upsert的特性构造了配置化的数据回溯任务，去回补可能因为Flink流处理任务运行失败而造成的缺失数据，其中的细节就不在这里展开了。

3.2 计算层面流批一体

对于计算层面流批一体的问题，上文提到希望寻找一个实现了Dataflow模型的计算引擎去统一处理批处理层和流处理层的数据计算，因此Flink就成为了最佳的技术选型。在我们原先的Lambda架构上使用Flink统一批处理层和流处理层的架构演变为下面这样：

图中绿色箭头是借助平台能力让Kafka数据每小时一次落到Hive中，可以看到批处理层存储使用Hive，流处理层存储使用Kafka。这里有一个关键点是Flink如何去访问Hive表？得益于Flink Hive Connector，我们可以实现这个功能。

3.3 存储及计算层面流批一体实践

上述两种对Lambda架构的改进分别只在存储或计算层面做了流和批的统一，而我们的最终目标是希望能够在存储及计算层面均实现流批一体，将整体优势最大化，也才能称之为真正的“流批一体实时湖仓”。因此我们在这个方面做了较多实践，并已经形成可落地可复制的经验。由此构建我们结合Flink和Iceberg建设的流批一体实时湖仓架构：

图中OLAP表示我们可以使用各种OLAP引擎查询Iceberg中的中间结果数据，ClickHouse表示为了用户对报表结果的多维分析查询方便将DWS层的Iceberg数据同步到ClickHouse。

要实际落地整个链路，其中关键的一环是如何运行Flink的批处理任务，以及这些批处理任务如何实现任务依赖调度能力。感谢Flink社区在批处理作业上的支持，目前已经可以运行Flink批处理任务，并且可以使用Oozie、Airflow等开源调度框架配置批处理任务的依赖关系。我们已经在小范围数据验证了整个链路，下面给出demo用以说明如何实现。由于Flink的API中最高层次的抽象是SQL，因此Flink SQL是门槛最低的开发方式，相比DataStream API屏蔽了很多底层的实现，也能满足大部分不需要精细操作的一般需求，同时也是社区在流批一体持续投入的API，所以我们选择Flink SQL作为具体实现的API。

demo需求：假设我们现有一张Iceberg表click在被Flink不断增量写入（DWD层明细表），表结构如下，其中click_date为分区字段：

click_date	string	点击日期
click_timestamp	bigint	点击时间戳
request_id	string	请求ID
app_id	bigint	游戏ID

我们有以下两个需求：

（1）按日统计每个游戏的点击量，我们希望这个结果是准确的

（2）按分钟统计每个游戏的点击量，我们希望这个结果的延时低

显然，这两个需求的主要计算逻辑是一致的，都是统计每个游戏的点击量，不同的是统计的时间窗口大小。对于需求（1），我们需要用批处理保证结果的准确性，对于需求（2），我们需要用流处理保证结果的实时性。在流批一体实时湖仓架构下，我们会如下开发这两个需求的代码。

对于需求（1），首先在Flink SQL中执行如下DDL创建source表：

CREATE TABLE source(
  app_id BIGINT,
  request_id STRING,
  click_date STRING
)
with (
  'connector' = 'iceberg',
  'catalog-name' = '%s',
  'catalog-database' = '%s',
  'catalog-table' = 'click',
  'cache-enabled' = 'false',
  'uri' = '%s',
  'warehouse' = '%s',
  'write.format.default' = 'PARQUET',
  'format-version' = '2'
)

其中%s填入对应的值即可，如catalog-database填入表click所在的Iceberg库等。同样执行DDL创建sink表的代码略去不表，只需要提前创建好对应的Iceberg表即可，假设sink表名称为click_batch。接下来我们执行如下Flink SQL即可通过批处理得到我们需要的结果数据：

insert overwrite click_batch
select '%s',app_id,count(distinct request_id) 
from source
where click_date='%s'
group by app_id

这里%s填的值从批处理作业外部传入，指定我们需要计算的是哪一天的数据，最终结果也写入下游Iceberg表对应的分区。需要注意的是打包好的JAR要指定运行时模式为"BATCH"，否则会因为执行模式非"BATCH"而不支持insert overwrite语法。我们对比了这个批处理作业的结果与使用Presto撰写相同逻辑查询ODS层表的结果，两者一致，说明结果是准确的。

对于需求（2），首先在Flink SQL中执行如下DDL创建source表：

CREATE TABLE source(
  app_id BIGINT,
  request_id STRING,
  click_timestamp BIGINT,
  ts as TO_TIMESTAMP(FROM_UNIXTIME(click_timestamp)),
  WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
)
 with (
  'connector' = 'iceberg',
  'catalog-name' = '%s',
  'catalog-database' = '%s',
  'catalog-table' = 'click',
  'cache-enabled' = 'false',
  'uri' = '%s',
  'warehouse' = '%s',
  'write.format.default' = 'PARQUET',
  'format-version' = '2'
)

这里我们指定了Watermark为click_timstamp - 1s。同样的，执行DDL创建sink表的代码略去不表，假设sink表名称为click_stream。接下来我们执行如下Flink SQL即可通过流处理得到我们需要的结果数据：

insert into click_stream
select
DATE_FORMAT(TUMBLE_START(ts, INTERVAL '60' SECOND),'yyyy-MM-dd HH:mm:ss'),app_id,count(distinct request_id)
from source /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s')*/
group by TUMBLE(ts, INTERVAL '60' SECOND),app_id

将打包好的JAR指定运行时模式为"STREAMING"，我们观察到结果数据的延时在分钟级，说明实时性是满足需求的。

将这个具体实践的结果对照我们进行流批一体实时湖仓建设前预设的目标，发现都已经达成了：

（1）存储层面流批一体，我们的批处理任务和流处理任务均是消费的同一张Iceberg表（此处为click表），不再需要两套存储系统支撑

（2）计算层面流批一体，我们的批处理任务和流处理任务在主要计算逻辑上复用了同一份Flink SQL代码，只是在“统计时间窗口”的处理上略有不同，这也是需求所决定而避免不了的，从而不再需要开发两套代码

在上述实践的过程中，我们也遇到了很多问题，通过自主思考以及积极与相关团队沟通，都得到了解决。在流批一体的实践中，分别在流处理，流转批及批处理中遇到了一个重要问题，下面分别对其给予介绍。

3.3.1 流式计算中数据保序问题

我们知道，在流式计算中窗口及定时器是底层操作，离开他们流式计算无从谈起。窗口关闭或者定时器触发是由Watermark决定的，Watermark机制其实是为了平衡结果数据的完整性和实时性，下面举一个简化的例子来说明。

假设我们现在需要每5s统计一次数据条数，也就是count(*)，同时我们设置Watermark=EventTime-1s，也就是水位值滞后事件时间1s，接下来有如下三条数据依次到达Flink内，我们看看会发生什么（图中蓝色虚线为Watermark，灰色背景为窗口）：

可以看到，当第一条事件时间为2s的数据到达时，Watermark为1s，此时[0s~5s)的窗口没有触发关闭，图中以虚线框表示。第二条事件时间为9s的数据到达时，Watermark上涨到了8s，此时[0s~5s)的窗口由于水位值大于等于窗口结束时间被触发计算并关闭，窗口中只有一条数据，因此输出的计算结果为1。等到第三条事件时间为4s的数据到达时，[0s~5s)的窗口已经在前面关闭输出结果了，因此这条本该属于[0s~5s)窗口的数据没有被计算在输出结果中，我们得到的计算结果是错误的。在这个例子中[0s~5s)的窗口统计数据条数正确结果应该是2，但是我们得到的结果是1。

如果我们在这个例子中希望得到正确的结果，就需要在事件时间为9s的数据到达时，[0s~5s)的窗口不会被触发计算和关闭，也就是Watermark<5s。假设我们设置Watermark=9s-Ns<5s，可以计算出N>4s，这个操作直接影响了我们看到流式计算结果的实时性。通过这个例子就能看出，Watermark相对事件时间所滞后的值其实需要平衡结果数据的完整性和实时性。

我们得到这个结论之后，其实就引申出了一个在流式计算中非常重要的问题——数据流入计算引擎时是需要保序的。也就是不能大量出现先产生的数据后被读到的情况，因为这种情况下，我们需要设置一个非常大的Watermark延迟，才能得到正确的计算结果，这样做就违背了我们希望流式计算尽快得到结果的初衷。以前我们使用消息队列存放流式计算的数据，消息队列FIFO（先进先出）的特性保证了不会出现大量数据乱序的问题，少数乱序数据可以通过调整Watermark策略去处理。但是现在我们消费的数据是放在Iceberg中的，Iceberg在设计时并没有特意考虑FIFO的特性，所以这个问题要怎么解决呢？

经过咨询相关团队同学，我们得知Iceberg可以通过修改代码支持FIFO的特性，当开启FIFO模式后Iceberg会一个个的消费Snapshot，并将读取的增量文件按照文件的写入时间排序后传递给下游。并且不仅在DataStream API支持了这个配置，Flink SQL也做了支持，只需要在创建source表的DDL中增加相应配置就能开启。

3.3.2 流转批问题

当我们思考批处理最上面一层的任务应该在什么时候启动时，其实就涉及一个流转批的问题。我们期望批处理的结果是准确的，也就是exactly-once的，那么必然需要保证批处理任务读取的数据不能有遗漏。在我们的架构中，批处理的DWS层数据依赖的DWD层数据是由Flink增量写入的，那么这里DWS层的批处理任务要什么时候启动，才能在保证所需要的数据全部落到DWD层的Iceberg后尽快运行呢？其实这个问题和Flink的窗口需要在什么时候触发计算是一样的，所以可以参考Flink中的Watermark机制去实现。我们先看看Flink的Watermark是如何更新和传递的：

图中的Flink任务具有四个输入分区和三个输出分区，可以看到这个任务事件时间时钟是类似于“木桶原理”一样，根据所有输入分区的Watermark中最小的一个来更新，并且在更新后广播到所有输出分区。我们可以参照这个方式将Iceberg的Watermark写入外部（比如Iceberg的元数据），然后启动一个任务不断轮询这个已经落到外部存储的Watermark，一旦发现最新的Watermark超过设定的阈值就让任务成功，从而让依赖这个轮询任务的下游批处理任务感知到它需要的全部数据已经到达。

同样地，经过和相关团队同学沟通后，我们得知Iceberg可以通过修改代码实现这个"Watermark"写入Iceberg元数据，并且可以将该配置写在Iceberg的表属性中。同时，我们可以封装好配套的Spark任务，配置检测的Watermark需要达到的值以及任务间隔，再配置下游批处理任务依赖这个checker任务就可以了。这样，批处理最上面一层的任务通过依赖checker任务就能被正确触发启动，从而保证了流转批的正确性。

3.3.3 关于批处理本质的思考

在实践Iceberg批处理任务的过程中，我们最开始使用的是Flink DataStream API来读写Iceberg的，但是当我们按之前的经验希望从Iceberg中读出某个分区的数据时，发现这一点却不是那么理所应当的容易做到。这启发了我们对批处理本质的思考，或者说我们希望批处理任务是怎么样的。

因为Iceberg Source可以指定以批的模式读取某一个snapshotId，但是这个snapshot中有表在那一时刻所有的数据，我们希望获取某个分区的数据，还需要在这个时刻所有数据中去按分区字段做过滤。随着表数据积累得越来越多，需要耗费越来越多的时间在读取IO和分区过滤上，并且将所有数据加载到计算框架内存中可能会造成OOM的问题，这显然是不合理的。那么有没有一种方法可以在存储层面就做了过滤，让计算框架只获取到某个分区的数据？经过调研我们发现社区Iceberg在0.11.0版本中对Flink做了深度集成，其中实现了Filter Pushdown这个功能，通过Filter Pushdown，数据可以在存储层就被过滤而无需从存储层读取全部数据，然后将数据给到计算引擎，这个功能显著地提升了数据获取的效率。我们咨询了相关团队同学后，得知Flink DataStream API需要在Iceberg Source中添加一个filters()，其中需要传入Expression去定义过滤规则，这种方式显得比较复杂。但是如果写Flink SQL，代码就显得很简洁了，只需要在where后指定分区字段的过滤条件，Iceberg侧就能将Filter下推到存储层做相应分区裁剪，如我们前文批处理demo中的：

where click_date='%s'

这样，结合上文提到的Iceberg Watermark checker，我们就在批处理的Source处将需要计算的数据不重不漏地获取到了，其实也可以理解为将需要的数据分组到了这个分区中。相应地，批处理在Sink处有没有什么需要注意的？回到Lambda架构中提出的将整个数据处理流程抽象成 View = Query(Data) 这个函数，现在我们在Source处保证了Data是exactly-once的，用户需要的View也不会变化，但是实际开发批处理任务中我们可能会因为各种bug反复修改Query()的逻辑，这就需要Sink处能够支持高效地将之前错误的数据删除，将新的正确数据插入，即overwrite的能力，也就是我们前文批处理demo中的：