阿里云数据湖统一元数据与存储管理实践

志明与数据 2023-05-26

603

导读随着数据湖的发展和日渐增长的需求，对数据湖进行统一元数据和存储管理也显得日趋重要。本文将分享阿里云在数据湖统一元数据与存储管理方面的实践。

主要内容包括以下四大部分：

1. 云上数据湖架构

2. 数据湖统一元数据

3. 数据湖存储管理与优化

4. 数据湖格式管理与优化

分享嘉宾｜范佚伦阿里云技术专家

编辑整理｜陈业利英祐科技

出品社区｜DataFun

云上数据湖架构

首先介绍一下数据湖相关的概念和架构。

不同的云产商对数据湖有着不同的定义。但是从关键词上来看，基本上都是围绕这几个特性和目标：

（1）统一存储，即数据湖是一个统一的中心化的数据存储；

（2）可以用来放一些原始数据；

（3）支持多种格式，包括结构化的数据和非结构化的数据。

首先，统一存储主要是为了解决数据孤岛的问题。因为传统的数据库或者是数据仓库在设计上是存算一体的，也就是在不同的查询引擎之间，数据需要经过清洗和同步。这样不管是在存储空间上，还是效率上，都存在一定的浪费。而数据湖上则是使用存算分离的查询引擎，典型的比如 Hadoop 生态的 Hive 和 Spark。再加上开放的存储格式，如 Parquet、ORC 等，来实现使用不同的引擎同时可以查询同一个数据的功能。这就是早期数据湖的架构。

在存储实现上，数据湖通常会使用扩展性比较高的，廉价的存储，比如 HDFS，或者云上的 OSS、S3 等对象存储。这样大家可以把更多的原始数据，非结构化数据直接放入，避免原始数据的丢失。

为了能够读取这些原始数据，计算引擎通常支持类似 schema on read 的方式，采取事后建模的高灵活性的解析方式，对数据格式没有很强的约束。这种灵活性也带来了一些弊端，比如其高度开放性可能导致对安全和权限的管理相比于数仓是有所差距的。另外，因为开放存储，并发写入的场景尤其是流式写入的场景，事务上对 ACID 的要求会更高。

是否有一种办法使我们既能够利用数据湖的优势，也能让数据湖拥有数仓的功能特性呢？

前两年 Databrick 提出了 Lakehouse 湖仓一体的概念，希望让数据湖能够实现更多数仓的企业级能力，让用户可以像使用数仓一样地使用数据湖。

Lakehouse 概念是在数据库的基础之上，添加了几层内容。

首先在存储上层，做了元数据的统一。对上层提供统一的元数据结构化 SQL 的接口，让不同的应用，可以使用相同的元数据访问数据。

另外在性能上，支持 cache，来优化数据湖读取性能。

并且，利用数据湖格式实现事务层。目前很火热的数据湖格式，Delta Lake、Hudi 和 Iceberg，使得我们现在提到数据湖场景，基本就跟这几个数据湖格式划等号了。虽然这有些夸张，但也足以证明它们在数据湖架构中的重要地位。

最后在底层的数据湖存储的实现上，相比于 HDFS，目前在云上也有使用对象存储作为数据湖存储的趋势。因为云上对象存储的扩展性相比于自建 HDFS 要高很多。不管是在成本上，还是在可用性上其实都是会高一些。

阿里云提供了一些产品功能来帮助用户使用数据湖的架构。

首先支持多引擎计算分析，比如常见的 EMR 的 Spark 和 Hive、Presto、StarRocks 这些引擎，以及阿里云自研的引擎，如 MaxCompute、Hologres，都可以进行湖上数据分析。可以根据不同场景来选择合适的引擎。

另一方面这些引擎为了能够无缝对接湖上的结构化数据，DLF（data lake formation）产品提供了统一的元数据和湖上的权限管理，作为整个 lakehouse 架构里的元数据管理层。在后面还会展开介绍。

最后在存储层上，云上的对象存储 OSS 是天生适合做数据湖存储的，并且成本不高。同时现在 OSS 也支持兼容 HDFS 接口的产品，OSS-HDFS，是完全支持 HDFS 接口的，更适合对接一些老版本的大数据引擎。

DLF 的核心能力是提供一个全托管的统一元数据服务，因为数据都已经放在数据湖上了，元数据需要一个中心化的管理才能实现多个引擎的无缝对接，这体现了元数据服务在数据湖里的重要性。这样不同引擎读写同一份数据是围绕统一的 schema 做操作的，而不是每个引擎都需要单独建外表。

同时围绕元数据，我们提供对数据的细粒度的权限管控。

另外也提供了数据湖上的一些存储管理的功能。

数据湖统一元数据

下面就来具体介绍阿里云数据湖的一个重要能力，数据湖上的统一元数据。

在开源大数据体系里，从早期的 map-reduce 到类似 SQL 查询语言 Hive 的诞生之后，Hive 逐渐成为了开源数仓的事实标准，围绕着 Hive 的元数据 Hive Metastore 也成为了对接开源数仓的元数据标准。从此以后各个引擎，包括 Spark、Presto 等都是支持对接 Hive Metastore，围绕 Hive Metastore 做元数据管理。

Hive Metastore 是一个常驻的无状态的服务，它可以部署一个或者多个实例。大数据引擎通过 thrift 协议连接 Hive Metastore 进行元数据的读写。

Hive Metastore 的元数据本身是需要存储到数据库上，通常会用 MySQL 作为 Hive Metastore 元数据的底层存储。

这就形成了常见的开源大数据元数据体系。

使用 Hive Metastore 管理元数据也存在着一些问题和挑战。

首先在功能层面上它是没有做多版本的，不能追溯之前的元数据版本。ACID 的特性和 LOCK 接口是和 Hive 引擎绑定的，在湖上多引擎的场景下，是没有办法利用到它的一些功能的。

另外因为它暴露的是 thrift 协议的接口，如果你自有服务，或者自研引擎需要去对接会相对麻烦一些。有时可能还需要直接连 MySQL 去读一些元数据，这也不是一个比较好的方法。

还有一个问题是它存在性能瓶颈，存在单点问题和运维成本，尤其是对元数据量比较大的用户，这是一个比较常见的问题。因为单点的 Hive Metastore Server 和 Metastore 后端连接的 MySQL 都可能会成为瓶颈，需要一些性能调优的工作。

上图中还列出了一些真实的客户问题。在 Hive Metastore 的使用过程中，首先会遇到的就是 JDBC 连接的问题，可能会遇到一些错误。比如有的时候我们查询元数据的所有请求都突然变慢了，这时首先要检查 MySQL 的状态，查看 MySQL 监控是否有慢 SQL。如果分区数总量很大的话，MySQL 表数量可能会达到上千万，会导致查询比较慢。这个时候，需要做一些数据清理，删除一些分区来缓解这个问题。另外在自建的数据管理系统或者外部系统中，通常不会用 thrift 协议去调用 Hive 的 Metastore Server，而是直连 JDBC，这样连接数多的话，也可能会带来一些额外的压力。

在内存方面，Hive Metastore Server 的内存存在 OOM 的风险。因为有些操作，比如 list partition，会加载全部分区对象，如果有人写了一个糟糕的查询，比如在一个很大的分区表上，没有加分区查询条件，就可能会拿到上百万的分区，最后导致整个 Hive Metastore 内存出现 full gc 或者 OOM 的情况，一旦 Hive Metastore 出问题，整个集群的作业都会受到影响。

列举几个我们遇到过的 StackoverflowError 的情况。如果 drop partition 的分区数量很多的话，在 Hive Metastore 的内部实现是递归的，可能会导致堆栈溢出报错，无法直接执行。

最后就是超时问题，因为 HMS 的客户端设计没有分页，是全量返回的。所以在拉取元数据的时候，可能会出现超时的情况，这也是一个风险点。

这些都是我们在使用 HMS 时候遇到的一些问题。

因此在云上，我们提供了全托管的元数据服务的 DLF（data lake formation），采用的是完全不同的架构，来解决上面大部分问题和痛点。

首先作为云产品，我们通过标准的 open API 暴露接口，提供了兼容 Hive2 和 Hive3 的 Metastore 接口的 client。这个 client 可以直接替换掉引擎的 Hive Metastore client 实现类，原本访问 Hive 元数据的地方可以直接替换为访问我们客户端的实现类，实现了无缝对接。

另外除了开源体系的引擎以外，我们也对接了阿里云上的其它大数据引擎，包括 Max Compute、Hologres、Flink 等等。云上其他大数据引擎也可以利用我们的统一元数据来进行元数据管理。这样真正做到了统一 catalog，用一个引擎写入，其它引擎读取。比如用 Flink 入湖，之后可以直接使用 Spark 查，再用 Hologres 等做 OLAP 分析，这些都可以直接采用同一个元数据来完成。

不同于 HMS 使用 MySQL，扩展性比较差，我们的元数据服务底层实现是用阿里云的表格存储。表格存储也是阿里云提供的一种服务，面向海量数据有非常强的伸缩能力，扩展性很高，所以不用担心分区数过大带来的扩展性问题。

因为我们是一个全托管的服务，对使用方提供 SLA，高可用保障，前面提到的运维问题也可以避免。

总结一下，我们的统一元数据的优势为，一方面因为是全托管，可以减少元数据运维成本；另一方面真正实现了对接云上多引擎。

再补充一些关于元数据本身实现的细节。

首先元数据的客户端是兼容 Hive Metastore 行为的，实现了 Hive Metastore 的接口，可以直接去对接 Hive 生态相关的大数据引擎。Hive Metastore 内部的有些行为，比如在创建 partition 的时候统计 table size 等动作，都会保留在客户端里，所以不用担心兼容性问题。

另外客户端会做一些性能优化，包括异常重试、并发读取、分页查询等。对于重复提交的请求，客户端也会做一些合并压缩，减少 IO 开销。

在服务内部，除了刚才提到的存储层的高扩展性以外，我们也通过一些自动的分区索引，再做一些分区过滤的性能提升。

总体来讲在元数据的性能上，在一些小表上可能跟 RDS 有些差距，但是并不明显。在大分区表上，比如单表有 300 万分区的场景下，我们的查询性能会有比较明显的优势。比如在 300 万分区表下，如果分区条件全部命中，list partition by filter 在我们的元数据可以在 0.5 秒内返回，但是在 RDS 上因为它的分区值没有索引，需要花 5 秒左右才能返回。

在元数据的功能上再举几个例子。

首先是元数据多版本，我们会记住元数据每一次更新的前后状态，可以看到什么时间点加了什么字段，是谁做的修改的。有比较好的回溯机制，实现元数据审计。在元数据检索上，我们的元数据本身会把内容同步到 ES 搜索引擎里，对外暴露，可以通过字段搜表，也可以做全局搜索。

再来看一下权限相关的问题。

在开源大数据场景下做用户级别的权限控制，通常会有这么几种方案：

Hive 本身提供的认证能力，storage-based authorization和sql-standard-based authorization。但是 Hive 的实现都是跟 Hive 引擎绑定的。通常在其它引擎是无法使用到它的功能的，基本上也没有人真正会在其它引擎上去使用。

大家通常做法是用 Ranger 来做权限管理。Ranger 是一个通用的多引擎方案，它可以对 SQL 进行权限管理，也可以对文件系统做权限管理。它的原理是从 LDAP 同步用户信息，提供 UI 供用户配置权限。在大数据引擎这一侧，可以添加各种插件，通过插件来实现权限的拦截和检查。Ranger 是目前一个可行的方案，但是在公有云上面对我们自研的大数据引擎，是没法直接对接的。另一方面虽然它包括了如 SparkSQL 等类插件，但是官方的支持并不好，更多还是需要自研一些插件，或者找第三方插件，整体部署起来并没有那么简单。

因此在权限这一块，DLF 统一元数据也提供了鉴权的能力。

权限控制默认是没有开启的，因为不一定所有用户都需要，但是用户可以按 catalog 级别进行开关。catalog 是基于 database 之上的一层管理模型，如果基于 catalog 设置权限之后，管理员就可以在控制台进行具体的授权操作。包括 database、table、column、function 这些粒度都可以进行授权。也可以设置不同 action 的权限，比如只给某个人对某个 table 设置 select 权限，而不设置 insert 的权限。同时也支持 RBAC，可以把权限包在 role 里，统一赋权给一组用户，这些基本的能力都是具备的。

在鉴权环节的实现上，我们提供了两个层面的鉴权，第一层面是元数据的 API，我想要查看 table 或者 create table，这种动作会在服务端上鉴权。因为我们的云服务会直接去鉴权，判断发送请求的用户角色是否有相应动作的权限，如果没有就会进行拦截。另外因为有些 SQL 操作在元数据层面感知不到，比如在元数据上可能就是查一张表，但是并不知道是在往里写数据还是在读数据，这个时候和 Ranger 类似，我们也提供了引擎的插件，可以放在 Spark、Hive 上做一层拦截。和 Ranger 类似，会在内部检查代理用户到底有没有 select 权限，没有的话去做拦截。这两层的鉴权模型，适用于不同的场景。

再介绍一个额外功能，就是元数据迁移。元数据本身无论在云上，还是自建的 MySQL 的元数据，如果想要迁移，都需要一个迁移的过程。为了简化这个过程，我们在产品上做了元数据迁移的功能，在控制台上就可以做数据迁移。

简单来讲我们会去连远端的 MySQL 数据库，如果这个数据库在阿里云 VPC 内，会自动打通网络，通过 JDBC 直接拉取元数据，转换成我们云上的 DLF 元数据，这是直接产品化的。除了导入需求，可能还会有导出需求，包括两边元数据对比的需求。这些也提供了现成的工具可以直接使用。在元数据迁移方面，不管是导入导出还是其它方面的需求，我们都保持开放性，不需要担心元数据被绑定的问题。

除了元数据迁移，可能在有些场景下还需要做元数据抽取，快速构建出湖上的元数据。元数据抽取适合于这样的场景，比如数据湖上已经有一些数据文件了，可能是从其它数仓拷贝过来的，或者是一些零散的 CSV 数据集文件等等。这个时候因为我们没有对应表的元数据，就需要用 DDL 语句自己去建表，再做查询，比较麻烦的，也容易出错。尤其是对于像 JSON 这种半结构化的嵌套类型，更难去写建表语句。这种情况下使用我们这个元数据抽取功能就比较方便，可以直接把元数据给推断出来。用户只需要填写 OSS 路径，我们会根据路径格式自动扫描下面的表，包括分区值，创建好之后，就会写入到元数据里进行直接查询了。包括 CSV、JSON、Parquet 、ORC 等各种格式，也包括湖格式都是可以识别出来的。值得注意的是因为我们做格式推断需要扫描所有数据，会比较耗时，于是我们采用了快速采样的方式。

数据湖存储管理与优化

接下来介绍我们在数据湖存储分流方面做的一些管理和优化。

首先介绍一下元仓，元仓是我们在元数据存储之外做的一个在线的元数据的数据仓库。因为元数据存储本身是在线服务，需要比较高的读写事务保障，有些后台分析，包括一些聚合查询是不适合在这里做的。于是我们做了一个实时的元数据仓库。元仓底层是基于 Max Compute 和 Hologres 实现的，它会收集元数据的变更信息，也会收集计算引擎的查询和写入的信息，包括存储上的信息都会实时收集到。这样我们就形成了围绕 database 的 table partition 做的指标库，即 data profile 指标。我们会把这些指标通过标准的 API 暴露出来。一方面可以在控制台上可以做统计分析，包括对接我们的一些云产品，如 dataworks 之类，做一些数据展示和预估。另一方面这些指标可以用来做存储生命周期的优化和管理。

接下来举例介绍一下 Data Profile 指标的几个实现。

首先是表和分区的大小，这是一个比较基础的属性。通常来讲，表和分区大小是写在元数据层，即 Hive 元数据的 table property 里面，本身就定义了，计算引擎会在创建表或者分区的时候写入。但是不同引擎写入的标准会不一样，比如 Hive 是叫 totalSize，Spark 是以 Spark 开头的属性值。另外，这些写入也需要一些参数去开启，不开启是不会进行写入的。所以在实际情况中会发现元数据本身存储的表大小是不准确的。

在元仓里，因为我们默认大部分数据湖使用的是 OSS，我们会通过 OSS 的底层存储来获取表分区的大小，这样可以最大限度保证数据的准确性。因为 OSS 提供了一个 t+1 更新的存储清单，这一点类似于 LAMBDA 架构，会 t+1 更新存储清单的表和分区的存储大小。另外对于实时表和分区的变更，我们也会监听到，再实时的从 OSS 那边拿到最新的大小去做更新。也就是通过存量加增量的流程去获取表分区的大小，拿到大小之后，会每天产出一些分析报表，比如表的存储排名，文件大小占比等等。因此我们可以看到哪些表，哪些分区的存储占用比较大，去做相应的优化。

上面是一个比较完整的湖上管理视图。

另外再介绍两个关键指标。

第一个指标是表和分区的访问频次，通过访问频次可以鉴别那些仍然在用但访问不频繁的表。这些表可以在 OSS 底层置为低频存储，照常读取的同时可以节省一些成本。在原理上我们通过使用引擎的 Hook 来实现对访问频次的获取，我们解析 SQL 的 plan，拿到它读取的表和分区，再提交到元数据服务里去做记录，最后把访问频次指标统计出来。

第二个指标是表和分区的最后访问时间。它可以用来识别这个表和分区是否还有人在访问。为了保证指标的准确性，最后访问时间是通过 OSS 底层的访问日志获取的。这样不管通过任何引擎任何途径读这里面的数据，访问时间都会获取到。最后对于没有人使用的表和分区，就可以考虑做归档或者删除。

结合这几个指标，更有利于我们做库表分区的生命周期管理。因为湖上生命周期管理也是一大重点，因为数仓是有存储分层的概念，但在数据湖上是没有一个比较完整的管理能力。我们目前就在做这方面相关的事情。

首先我们使用的标准型 OSS 对象存储是提供了存储分层能力的，也可以按需设置成低频归档，冷归档这些层次。设置好归档之后，会对数据访问方式产生影响，但是存储成本会大幅降低。

用户首先可以设置一些规则，包括基于分区值，分区的创建时间，上面提到的访问频次等指标，配置规则设定阈值，比如多长时间没人访问或者 30 天内访问频次低于几次。后台就会定期把符合这些条件的分区的整个目录做归档，或置为低频等。

另外归档和冷归档做了之后是不能直接访问的，需要一个解冻的流程。如果用户有一天需要访问已经归档的数据，可以一键解冻，整个目录就可以直接使用，而不需要像 OSS 那样逐个文件进行解冻操作。这种存储生命周期管理的存储优化，对于存储量比较高的数据湖用户来说会是一个比较好的实践。

数据湖格式管理与优化

最后介绍一下在数据湖格式层面，我们做的管理和优化。

常见湖格式 Hudi、Iceberg 有几个特点，为了实现 ACID，它们的底层数据文件更新，copy on write 之后，读取的都是新版本的数据文件，但是老版本的数据还会保留在存储侧。时间一长就需要清理历史版本的数据文件。另一方面频繁流式写入会产生很多小文件，通常可以使用命令手动清理，或者结合在 streamming 任务当中，配置一些参数，比如多少 commit 清理一次，但是这对流式写入本身的性能会产生影响。针对这种情况，业内很多公司都使用额外部署 table service 的方式，不影响流式写入，另起一个批作业去清理和优化。DLF 把这种 table service 做在了云服务里面，这样使用 DLF 湖格式的用户，可以直接在控制台上配置规则，比如基于版本号更新多少次就做一次清理。后台就会跑任务做 vacuum 或者 optimize 命令，整个过程也是全托管的，用户不用关心背后使用的资源。

具体实现原理为，元仓会维护很多元数据的变化和引擎消息，也会感知到哪些湖格式表发生了写入和变化。每一次表的写入，就会触发规则引擎去做一次判断是否满足条件，如果满足条件就会触发动作的执行。目前我们对 Delta Lake 已经有比较完整的支持了，对 Hudi 的支持也在进行当中。这是一个比较新的模块。

再具体介绍下湖格式管理的几种优化策略。

第一种也是最常见的，基于版本间隔，清理清理历史文件或者合并小文件。比如写入了 20 个 commit 之后就会自动触发整个表的清理，或者小文件合并。这个阈值是可以随用户级别或者作业级别做配置的。内部会把这些合并的任务放在一个队列里，这样前一个合并任务还没有跑完，是不会跑下一个合并任务的，避免并发执行，产生写冲突现象。

第二种合并规则是我们在客户实践过程当中觉得比较实用的，基于时间分区自动合并上一个分区的小文件。因为在流式写入的场景下，通常会按时间顺序去命名分区值，每写入一个新分区就代表上一个分区写入停止。在这个时候，一旦发现有新分区创建，就可以去对上一个分区做一些优化和合并的动作。这样上一个分区后续的查询性能就能得到保证，同时这种做法也能最大程度避免合并任务和写入流任务的写冲突。当然为了实现这个方案，我们内部也是做了时间格式的支持，自动处理了很多分区值的时间格式。这样我们就可以自动识别这些时间分区哪个分区是最新的，哪个分区是上一分区的。

问答环节

Q1：DLF 元数据的管理，跟 Databricks 推的 unity catalog 有什么区别？

A1：DLF 元数据管理有点类似于 Hive Metastore 的升级。Databricks 推的 unity catalog 其实是跟它的执行引擎，Databricks 的 Spark 的绑定比较多，它是基于 Databricks 的引擎去做很多事情。我们对单个引擎的集成没有 unity catalog 那么完整，但是更聚焦在云上的统一元数据，即同一份元数据可以被云上各种各样的引擎，包括自研的和开源的引擎，都能统一的进行读写。总结：我们对云上统一的数据这个角度做的比较多，针对的是多引擎的打通。对某一个引擎内部做的集成没有 unity catalog 那么深入。

Q2：DLF 的 OPEN API 是开源的吗？

A2：首先我们是一个全托管的云产品，内部的实现是做成云服务。然后我们会提供标准的 API，用户可以通过阿里云的 SDK 对 API 的调用和使用。最后我们的元数据 client，适配 Hive client，同时 client 本身也是开源的，内部的元数据服务是在云上实现的。

Q3：DLF 针对小文件治理，计算资源控制。跟湖格式相关的小文件合并的问题。

A3：目前因为我们的湖格式的小文件治理产品还处在公测阶段，还没有进行真正的计费。底层的资源我们是内部提供的，不使用用户的资源。我们内部是会做一些针对单租户的，最大的使用量的限制的。目前计费策略还没有明确的推出。这个可能会等到后续足够完善之后再去做相关事情。

Q4：现在的 Hive Hook 解析 HSQ 的 SQL，Matestore 的 listener 能监听 DDL 吗？

A4：我们现在实现的 listener 是能够监听到 DDL 的。首先 DLF 元数据本身，因为刚才提到了我们也有元仓。其实内部是会监听到所有元数据的变更，同时我们也会基于引擎的 Hook 去监听表查询的信息，维护到 DLF 元仓里面。因为我们的实现是没有 Metastore 的，用户可以从 DLF 的 Data Profile 的 API 进行获取。如果想自己实现像以前的 Hive Metastore 一样的 Metastore 的 listener，这是不支持的。但是可以基于我们云上的 API 去获取元信息。

今天的分享就到这里，谢谢大家。