海量存储的批量计算框架
导读
本文介绍了百度针对海量存储数据计算需求研发的HTAP表格存储系统及计算调度架构。项目背景源于原有存储系统难以满足日益增长的OLAP业务需求，因此构建了集OLTP与OLAP于一体的HTAP系统，通过存算分离、Serverless设计等创新点提升IO访问能力和资源利用率。同时，自研的计算与调度系统实现了任务开发的SQL化和数据处理的FaaS化，简化了业务使用成本，提高了开发效率。整体方案在存储成本、IO能力、IO放大率等方面取得显著成果，为海量存储数据的计算提供了高效、灵活的解决方案。
01 项目背景及目标
1.1 项目背景
搜索内容存储团队主要负责各类数据，如网页、图片、网页关系等，的在线存储读写(OLTP)、离线高吞吐计算(OLAP)等工作。
原有架构底层存储系统普通采用百度自研表格存储（Table）来完成数据的读、写、存工作，此存储系统更偏向于OLTP业务场景。随着近几年大数据计算、AI模型训练的演进，对存储系统OLAP业务场景的依赖越来越重，如数据关系分析、全网数据分析、AI样本数据管理筛选。在OLTP存储场景的架构下，支持OLAP存储需求对资源成本、系统吞吐、业务时效带来了巨大挑战。为此我们在百度自研表格存储之外，结合业务实际workflow针对性优化，增加构建了一套符合业务需求的HTAP表格存储系统以及相应的计算框架，共同组成面向海量存储数据的大批量计算架构系统。
1.2 项目目标
-
提供海量存储数据计算的超高IO访问能力。当前内容存储数据达几十P+，访问频率按照每周一轮估算，平均IO能力需要达到34G/s，峰值IO能力需要达到200G/s。面对如此庞大的IO访问能力，需要从文件系统、存储引擎、分布式存储系统、访问模型等全方位进行深度优化来满足需求；
-
提供海量存储数据计算的快速开发&部署能力。在提供海量访问能力的同时，也需要为业务提供访问配套的基础设施，来满足业务开发&部署计算任务的需求。
02 现有研发条件和工作基础
搜索内容架构存储团负责各类数据，如网页、图片、网页关系等，的在线存储读写(OLTP)、离线高吞吐计算(OLAP)等工作。面对当前海量存储数据的计算需求有清晰的技术和业务认知，第一视角明确清楚地知道系统瓶颈、技术难点、业务需求。
-
系统瓶颈——当前存储系统能提供的IO能力与业务计算需求之间的矛盾。随着大数据、机器学习、大语言模型等新技术的兴起，业务对数据的计算访问需求越来越强烈，然而存储系统的IO能力却一直止步不前。为此，迫切需要一款面向数据计算的存储系统；
-
技术难点——数据表格存储系统的数据访问模型与计算模型之间的矛盾。当前架构底层存储普遍采用百度自研表格存储(Table)来完成数据的读写存工作，此存储系统更偏向于OLTP业务场景。但随着近几年大数据计算、AI模型训练的演进，对存储系统OLAP业务场景的依赖越来越重，如数据关系分析、全网数据分析、AI样本数据管理筛选。在OLTP存储场景的架构下，支持OLAP存储需求对资源成本、系统吞吐、业务时效带来了巨大技术挑战；
-
业务需求——方便高效快速的任务开发&部署能力的需要。在大量搜索内容OLAP workflow中，从表格存储系统中提取筛选数据只占全部任务的一小部分，大量任务需要对数据进行加工处理得到需要的结果。常规的做法是多任务串联，这样做的缺陷是大量中间临时数据存储开销。为此我们为HTAP表格存储系统构建了一套计算与调度系统。
03 整体方案
3.1 概览
本项目拟研发面向海量存储数据的大批量计算架构，主要分为两大系统，HTAP表格存储系统、计算&调度架构。
3.1.1 HTAP表格存储系统
△图2.3
-
架构采用业界HTAP主流设计思想，将OLTP和OLAP workflow拆分到两套存储系统中，如F1 Lightning、ByteHTAP，在SDK层根据任务类型分发到不同的存储系统中；
-
OLTP存储系统——Neptune，采用Multi-Raft分布式协议组建存储集群，采用本地磁盘(SSD/HDD等) + 百度分布式文件系统AFS组成存储介质；
-
OLAP存储系统——Saturn，Serverless设计模式，无常驻Server，即用即加载，贴合OLAP workflow的不确定性和间歇性；
-
OLTP与OLAP存储系统间，采用数据文件硬链的方式进行数据同步，全版本替换，成本低、速度快，充分贴合Saturn Serverless设计模式。
如上架构设计图，可将OLTP与OLAP workflow拆分到两套独立的系统中，解决上述提到的存算耦合问题。
- 解决存储空间放大问题。空间放大主要带来的问题是存储节点成本，Workflow分离的架构将OLAP需要的数据文件采用AFS低成本存储，减少了对存储节点存储空间的压力。
△图2.4
OLAP存储系统的数据写入并没有使用常见的log redo或raft learner模式，最主要还是在保证OLAP存储系统的Serverless特性的同时，又能实时感知到OLTP系统的最新写入结果。
- 解决存储节点资源冗余问题。拆分后，分布式存储节点将大量重型OLAP workflow转移到OLAP存储——Saturn中，将极大减少存储节点的计算压力。同时，OLAP存储的Serverless设计模式又可贴合workflow的不确定性和间歇性。
△图2.5 Saturn Serverless模型
计算节点可以部署在任意计算集群中，如Map-Reduce、自研计算节点Pioneer等，在SDK中直接初始化存储引擎，从AFS中访问对应分片的数据文件。计算节点可充分利用云原生系统（PaaS）的弹性资源，解决资源常驻冗余问题。
3.1.2 一次开发，多端部署
-
任务生成。自研KQL数据查询语言。在任务生成阶段将KQL语句解析优化成相关的调度任务，一个Job包含多个Task。
-
任务调度。
-
任务调度的计算节点可以是Map-Reduce，也可以是自研计算集群Pioneer，负责不同计算场景。
-
任务运行容器负责数据依赖部署和运行计算框架。
-
计算框架采用插件化设计思想，依托KQL语言进行差异化描述。计算框架的最大特点是，可在数据处理节点执行用户自定义FaaS函数。
3.2 详细介绍
3.2.1 HTAP表格存储系统
3.2.1.1 OLTP存储系统——Neptune
Neptune引擎主要支持四类操作：写、删、读、Scan。每一类操作都通过RegionMapper进行映射，对外隔离分区概念。
Neptune存在两类分区：索引分区、数据分区。
-
索引分区。索引分区用于减少因为数据分区导致Key所在数据分区不明确导致的随机访问IO放大问题，提升随机查性能。
-
数据分区。Neptune可配置多个数据分区，每个数据分区内包含多个Locality-Group。分区间的数据理论上是互斥的。
Neptune各类操作的流程：
-
写操作：
-
根据RowWriter中设置的Region信息找到需要写入的Region的Handle，按照列语义将数据序列化成RawData。
-
同时根据Region信息生成当前Key的Region索引信息。
-
将RawData与RegionIndex作为一条操作记录Commit到引擎中，整个操作为原子操作。
-
删操作：
-
由于存在Region的概念，删除某个Key是需要明确当前Key所在的分区。目前的做法是查询一遍分区索引获取分区信息，再准确删除对应分区的数据。这样带来一个问题，删除操作会增加一次分区查询操作，我们可以考虑将分区信息全部加载到内存提升性能。
-
读操作：
-
读操作类似删除操作，会首先查询分区索引表，如果在分区索引中查询不到则表明当前Key不存在，直接返回NotFound。否则，根据分区索引查询对应的分区即可。
-
Scan操作：
-
Scan时业务可以指定对应的分区以及CF信息，RegionMapper根据这些信息Select出合适的物理存储Handle，然后对这些物理存储进行Scan。
3.2.1.2 OLAP存储系统——Saturn
Saturn主要分三层：文件系统（File-System）、Table（表级别的抽象，非TG的Table）、访问层（SDK），Meta-Server为每一层提供全局Meta信息支持。
-
文件系统。Saturn既可以支持AFS，也支持本地文件系统，同时后续可以支持其他类型的文件系统。文件系统的类型对于Saturn来说是插件化可插拔的。使用AFS作为文件系统相比于Table在成本层面有巨大优势。
-
Table。一个抽象的Table包含多个Slice，理论上每个Slice间的数据是互斥的，这里引入数据模型的概念。当前支持两种数据模型：哈希序(hash order)、全局序(global order)，两种模型与Table完全对等。
-
SDK。SDK目前支持Seek和Scan功能，使用方式跟通用的列存储系统保持一致，SDK直接与文件系统（AFS）连接，对外提供存储Serverless的访问能力。
同时，Table数据的更新和构建包含两种模式：全量构建、增量合并。
-
全量构建。全量构建通过完整Dump Table数据的方式对表中的每个分片进行逐步替换，替换过程中采用多版本机制保证访问的稳定性。
-
增量合并。增量合并通过控制TG Table做Major Compaction的时机，保证每次获取增量数据前不会发生Major Compaction。增量数据通过Snapshot的形式对外提供所有的操作记录，这些记录保存在Table SST文件中，Saturn把这些SST文件Transform成自身协议的SST，再发起Ingest操作即可。
3.2.1.3 存储引擎优化——数据行分区
数据行分区思想在很多OLAP存储系统中很常见，如当前比较流行的一些数据湖架构，ClickHouse、IceBerg等。在表格存储中，数据行分区的好处是可以极大减少在数据行筛选过程中IO放大率。以下是我们在存储引擎中支持数据行分区的设计思路：
△图2.6
数据行分区的思想在OLTP和OLAP存储引擎中都有使用，OLTP存储引擎以数据行分区构建的数据文件可直接被OLAP存储引擎加载，减少了OLAP存储的数据构建工作。
数据行分区在Write、Read、Scan场景下的处理流程分别为：
-
Write操作。Write时会根据请求中的特殊Region描述，如分区键，找到需要写入的Region-Index和Region上下文，前者保存Key的分区索引信息，后者中保存实际数据，操作记录由WAL中保存。
-
Read操作。Read操作相比通常直接访问数据，需要多进行一次分区索引访问，为减少多一次访问带来的性能折损，我们将分区索引信息全内存化。由于索引数据非常小，因此全内存化是可接受的。
-
Scan操作。Scan操作相比之下没有任何变更，但在Scan特殊分区场景下可大量减少IO放大。因为相比之前的行过滤模式，可直接跳过大量不需要的数据。
在业务存储支持时，合理设置数据行分区，可极大减少数据行筛选过程中的IO放大率。
3.2.1.4 存储引擎优化——增量数据筛选
在实际业务中，有很大一个场景是获取近期（如近几个小时、近一天）有值变化的数据，常规的做法是Scan全量数据，以时间区间作为过滤条件，筛选出符合条件的结果。但如此的筛选逻辑会带来严重的IO放大，因为满足条件的结果只占全量结果的一小部分。为此，我们在引擎层调整优化Compaction时机以及调整筛选流程，减少增量数据筛选过程中需要访问的数据文件集合，降低IO放大，业务提速。
△图2.7 LSMT
3.2.1.5 存储引擎优化——动态列结构
在OLAP存储引擎中，还存在一类访问场景会带来IO放大问题，数据列筛选。在表格存储系统中，一个Key可以包含多个列族（Column Family），一个列族中可以包含任何多个数据字段，这些字段以行结构存储在同一物理存储（Locality Group）中，当筛选特定数据列时，需要进行整行读取，然后过滤出需要的字段，这也将带来IO放大问题。
同时，OLAP workflow的访问不确定性导致存储层无法及时调整数据在物理存储中的结构。为此，我们引入动态列结构的概念，在逻辑层对业务透明，在物理层根据近期OLAP workflow特性及时调整物理结构。
△图2.8
如上图，在逻辑存储中，分为两个LG，根据workflow特性，把业务常用的访问字段在Compaction阶段存放在同一物理存储结构中，反之，这样可以减少字段筛选阶段的IO放大率。
动态列结构只在OLAP存储引擎中生效，我们在原有OLAP存储中引入workflow收集以及compaction任务，将从OLTP存储中同步的数据构建成更适合OLAP场景的存储结构。
3.2.2 计算与调度架构
在本节，我们将介绍在此HTAP表格存储系统基础上，如何设计实现任务计算和调度系统，简化业务使用成本，提升业务效率。
在大量搜索内容OLAP workflow中，从表格存储系统中提取筛选数据只占全部任务的一小部分，大量任务需要对数据进行加工处理得到需要的结果。常规的做法是多任务串联，这样做的缺陷是大量中间临时数据存储开销。
为此我们为HTAP表格存储系统构建了一套计算与调度系统，系统两大特点：任务开发SQL化、数据处理FaaS化。
3.2.2.1 SQL化与FaaS化
我们充分贴合上述存储系统特性，自研了一套数据查询语言——KQL，KQL类似于SQL Server语法。同时，又结合存储系统特性以及计算框架，支持一些特殊语言能力，最主要的是能支持原生FaaS函数定义，当然也支持外部FaaS函数包依赖。
如下是一段KQL语句例子以及说明：
function classify = { #定义一个Python FaaS函数
def classify(cbytes, ids):
unique_ids=set(ids)
classify=int.from_bytes(cbytes, byteorder='little', signed=False)
while classify != 0:
tmp = classify & 0xFF
if tmp in unique_ids:
return True
classify = classify >> 8
return False
}
declare ids = [2, 8];
declare ts_end = function@gettimeofday_us(); # 调用Native Function获取时间
declare ts_beg = @ts_end - 24 * 3600 * 1000000; # 四则运算
select * from my_table region in timeliness # 利用存储分区特性，从my_table中的timeliness分区获取数据
where timestamp between @ts_beg and @ts_end # 利用存储增量区间特性，筛选增量数据
filter by function@classify(@cf0:types, @ids) # 在Filter阶段调用自定义FaaS函数
convert by json outlet by row;
desc: # 对计算框架进行特殊描述
--multi_output=true;
3.2.2.2 任务生成与调度
任务生成与调度主要分为三层，任务解析层、任务调度执行层、任务执行容器。
-
任务解析层。负责将KQL表达式解析成实际的任务执行计划，并保存在任务存储容器中。
-
任务调度执行层。负责将任务计划分发到任务执行容器，并轮训检测任务状态，执行探活、重试等操作。
-
任务执行容器。提供两种任务执行容器，Pioneer、EMR。前者为自研任务执行容器，后者为公司Map-Reduce执行平台。
3.3 技术经济指标
通过上述的架构设计以及优化手段，我们在IO能力、访问成本、开发效率等方面取得显著成果。
04 主要创新点
4.1 自研HTAP表格存储系统
结合业务特性以及实际需求，构建符合业务场景的HTAP存储系统。架构采用业界HTAP主流设计思想，将OLTP和OLAP workflow拆分到两套存储系统中，如F1 Lightning、ByteHTAP，在SDK层根据任务类型分发到不同的存储系统中。系统创新点如下：
-
存算分离架构。解决OLTP存储系统的空间放大问题，将OLAP Workflow从OLTP存储中分离，分离的架构将OLAP需要的数据文件采用AFS低成本存储，减少了对存储节点存储空间的压力。
-
OLAP Serverless设计。分布式存储节点将大量重型OLAP workflow转移到OLAP存储——Saturn中，将极大减少存储节点的计算压力。同时，OLAP存储的Serverless设计模式又可贴合workflow的不确定性和间歇性。计算节点可以部署在任意计算集群中，如Map-Reduce、自研计算节点Pioneer等，在SDK中直接初始化存储引擎，从AFS中访问对应分片的数据文件。计算节点可充分利用云原生系统（PaaS）的弹性资源，解决资源常驻冗余问题。
-
表格数据行分区。数据行分区思想在很多OLAP存储系统中很常见，如当前比较流行的一些数据湖架构，ClickHouse、IceBerg等。在表格存储中，数据行分区的好处是可以极大减少在数据行筛选过程中IO放大率。
-
增量数据筛选支持。在实际业务中，有很大一个场景是获取近期（如近几个小时、近一天）有值变化的数据，常规的做法是Scan全量数据，以时间区间作为过滤条件，筛选出符合条件的结果。但如此的筛选逻辑会带来严重的IO放大，因为满足条件的结果只占全量结果的一小部分。为此，我们在引擎层调整优化Compaction时机以及调整筛选流程，减少增量数据筛选过程中需要访问的数据文件集合，降低IO放大，业务提速。
-
表格数据动态列结构。根据workflow特性，把业务常用的访问字段在Compaction阶段存放在同一物理存储结构中，反之，这样可以减少字段筛选阶段的IO放大率。动态列结构只在OLAP存储引擎中生效，我们在原有OLAP存储中引入workflow收集以及compaction任务，将从OLTP存储中同步的数据构建成更适合OLAP场景的存储结构。
4.2 自研任务生成与调度系统
在大量搜索内容OLAP workflow中，从表格存储系统中提取筛选数据只占全部任务的一小部分，大量任务需要对数据进行加工处理得到需要的结果。常规的做法是多任务串联，这样做的缺陷是大量中间临时数据存储开销。
为此我们为HTAP表格存储系统构建了一套计算与调度系统，系统两大特点：任务开发SQL化、数据处理FaaS化。
-
SQL化。我们充分贴合上述存储系统特性，自研了一套数据查询语言——KQL，KQL类似于SQL Server语法。
-
FaaS化。在SQL化的基础上，同时结合存储系统特性以及计算框架，支持原生FaaS函数定义能力，当然也支持外部FaaS函数包依赖。
————END————
推荐阅读
