本节我们讨论的是字节实时数仓场景的初探以及遇到的问题和解决方案。

坦白地讲，在最初落地时大家对数据湖能支持线上生产的态度都是存疑的，我们开始的方案也就比较保守。我们首先挑选一些对比现有解决方案，数据湖具有凸显的优势的场景，针对其中的一些痛点问题尝试小规模的落地。

离线数仓有两个比较大的问题：

对于这样的场景，数据湖兼具时效性和高效更新能力。同时相对于实时数仓来说，数据湖可以一份存储，批流两用，从而直接进行高效的数据分析。

基于以上对业务的分析，我们会按照以下步骤来做一线的落地。

看上图我们原有的方案有三个Hive表，Hive Table 1，2，3。对于整个链路来说我们会把左边MySQL数据源的数据导到Table 1中，右边Redis的数据导到Table 2中，然后将两个表做Join。这里存在两个比较大的问题：

我们通过引入Hudi把天级的Dump分摊到每个小时进行Upsert。由于Hudi自身可以支持去重的逻辑，我们可以将Table 1看成一个实时的全量数据，当小时级别（例如23点）的数据一旦Upsert完成之后，我们就可以直接进行下游的Join逻辑，这样的话我们可以将数据的就绪时间提前3.5个小时左右，高峰期的资源消耗可以减少40%。

对于实时场景来说，当实时任务进行一个比较频繁的变更，比如优化或者新增指标的改动，一般需要校验实时任务的产出是否符合预期。我们当前的方案是会跑一个小时级别的Job，将一个小时的数据从Kafka Dump到Hive之后再校验全量数据是否符合预期。在一些比较紧急的场景下，我们只能抽查部分数据，这时候就对时效性的要求就比较高。在使用基于的Hudi 方案后，我们可以通过Flink将数据直接Upsert到Hudi表中，之后直接通过Presto查询数据从而做到全量数据近实时的可见可测。从线上效果来看可以极大提高实时任务的开发效率，同时保证数据质量。

在以上探索过程中遇到了比较多的问题，第一个问题就是易用性比较差，运维成本和解释成本比较高。对于易用性这一部分，我们起初是通过脚本来提交SQL，可以看到SQL中的参数是比较多的，并且包含DDL的Schema，这在当列数比较多的情况下是比较麻烦的，会导致易用性较差，并且对业务侧来说也是不可接受的。

对于以上问题我们做了一个针对性的解决方案，首先我们对之前的任务提交方式替换为了纯SQL化提交，并且通过接入统一的Catalog自动化读取 Schema和必要参数，入湖的SQL就可以简化为如图的形式。

接下来让我们看字节目前基于Hudi的实时数仓整体链路。

可以看到，我们支持数据的实时入湖，例如MySQL，Kafka通过Flink可以直接落到Hudi；也支持进行一定的湖内计算，比如图中左下将MySQL数据通过Flink导入Hudi进一步通过Flink做一些计算后再落到Hudi。在数据分析方面，我们可以使用Spark和Presto连接看板BI进行一些交互式查询。当我们需要接到其他在线系统，尤其是QPS较高的场景，我们会先接入到KV存储，再接入业务系统。

让我们来看具体场景。

对于一个实时多维汇总的场景，我们可以把Kafka 数据增量写入到 Hudi 的轻度汇总层中。对于分析场景，可以基于 Presto 按需进行多维度的重度汇总计算，并可以直接构建对应的可视化看板。这个场景对QPS和延迟要求都不是很高，所以可以直接构建，但是对于高 QPS 和低延迟诉求的数据产品场景，目前的一个解决方案是通过 Presto 进行多维度预计算，然后导入到 KV 系统，进一步对接数据产品。从中长期来看我们会采取基于物化视图的方式，这样就可以进一步去简化业务侧的一些操作。

在以上链路中，我们也遇到了比较多的问题：

面对这些问题，我接下来简单介绍一下针对性的一些解决方案：

1）写入稳定性治理

这一块我们通过异步的Compaction + Compaction Service的方案去解决这个问题。我们之前Flink入湖默认是在Flink内部去做Compaction，发现这一步是暴露以上一系列问题的关键。经过优化，Flink入湖任务只负责增量数据的写入，以及 Schedule Compaction逻辑，而Compaction执行则由Compaction Service负责。具体而言，Compaction Service 会从Hudi Metastore异步拉取Pending Compaction Plan，并提交Spark批任务完成实际的Compact。Compaction执行任务与Flink写入任务完全异步隔离，从而对稳定性有较大提升。

2）高效更新索引

支持数据量级的大幅提升。简单来说，我们可以基于哈希计算快速定位目标文件，提升写入性能；同时可以进行哈希过滤，从而也可以进行查询分析侧的优化。

3）请求模型的优化

当前的Hudi社区版的WriteTask 会轮询Timeline，导致持续访问Hudi Metastore，从而造成拓展能力受限的问题。我们将WriteTask的轮询请求从Hudi Metastore转移到了对JobManager缓存的拉取，这样就能大幅降低对Hudi Metastore的影响。经过这个优化可以让我们从几十万量级的RPS(Request Per Sec)提升到近千万的量级。

接下来我们来讲一下查询相关的优化。

4）MergeOnRead列裁剪

对于原生的MergeOnRead来说，我们会在全量读取LogFile和BaseFile之后做合并，这在只查询部分列的时候会造成性能损耗，尤其是列比较多的情况。我们所做的优化是把列的读取下推到Scan层，同时在进行log文件合并时，会使用map结构存储K，V（K是主键，V是行记录），之后对行记录做列裁剪，最后再进行Log Merge的操作。这样会对序列化和反序列化开销以及内存使用率都有极大降低。

5）并行读优化

一般引擎层在读Hudi时，一个Filegroup只对应一个Task，这样当单个 FileGroup 数据量较大时就极易造成性能瓶颈。我们对此的优化方案是对BaseFile进行切分，每个切分的文件对应一个Task从而提高读并行度。

6）Combine Engine

Hudi社区版目前在内存中对数据的合并和传输的实现完全是基于Avro格式，这会造成与具体引擎对接时有大量的序列化与反序列化计算，从而导致比较大的性能问题。对于这个问题我们与社区合作做了Combine Engine的优化，具体做法就是将接口深入到了引擎层的数据结构。例如在读取FileGroup时我们直接读取的就是Spark的InternalRow或是Flink的RowData，从而尽量减少对Avro格式的依赖。这样的优化可以极大地提高MergeOnRead和Compaction的性能。