Daft 对接 Apache Gravitino：统一元数据目录接入、多格式表读取与 gvfs:// 文件协议实战

Daft 对接 Apache Gravitino统一元数据目录接入、多格式表读取与 gvfs:// 文件协议实战【免费下载链接】DaftHigh-performance data engine for AI and multimodal workloads. Process images, audio, video, and structured data at any scale项目地址: https://gitcode.com/GitHub_Trending/da/DaftApache Gravitino 是一个开源统一数据目录为 Iceberg、Hive 等表以及 S3/GCS/Azure Blob 上的 fileset原始文件集合提供统一的元数据管理。Daft 通过内置的 Gravitino 连接器API 目前处于 beta 阶段可能随连接器演进而变化让你可以直接用Catalog.from_gravitino()导航目录、读取多格式表并通过gvfs://协议对 Gravitino fileset 进行透明的文件读写。读完本文你可以掌握如何在 Daft 中连接 Gravitino 并完成认证配置、如何通过目录对象读取 Iceberg 与 Hive/Parquet 表、gvfs://URL 的完整读写用法以及连接器在 Python 客户端与 Rust IO 层的实际实现原理。安装与前置要求使用 Gravitino 连接器时需要以gravitino选项安装 Daftpip install daft[gravitino]从 pyproject.toml 可以看到该 extras 实际引入的依赖是requests2.28.0,3.0.0——因为当前版本直接调用 Gravitino 的 REST API而不是使用 Gravitino 官方 Python client。其他运行前提Apache Gravitino 服务端0.9.0 及以上版本Pythonrequests库随上述 extras 一并安装云存储凭据需在 Gravitino 侧配置好对应的存储凭据例如 S3 的访问密钥。注意文档明确标注这些 API 处于 beta 阶段接口在连接器开发过程中可能发生变化。连接器功能总览Daft 的 Gravitino 连接器提供以下能力Catalog Navigation目录导航列举 catalogs、schemas 和 tablesMulti-Format Tables多格式表通过catalog.get_table(...).read()读取 Iceberg 表以及 Hive/Parquet 表Table Management表管理加载已有表Fileset Support访问 Gravitino fileset 管理的原始文件GVFS Protocol使用gvfs://URL 直接读写 fileset 中的文件Authentication支持 simple用户名/口令和 OAuth2Bearer Token两种认证方式Daft Catalog Integration通过Catalog.from_gravitino()与 Daft 的目录系统打通。连接 GravitinoCatalog.from_gravitino最核心的入口是Catalog.from_gravitino()它由 daft/catalog/init.py 定义内部委托给daft.catalog.__gravitino.load_gravitino()。完整参数如下auth_type默认simple其余凭据参数可选参数类型说明endpointstrGravitino 服务端点 URL如http://localhost:8090metalake_namestr要连接的 metalake 名称auth_typesimple \| oauth2认证类型默认simpleusernamestr \| Nonesimple 认证使用的用户名passwordstr \| Nonesimple 认证使用的密码tokenstr \| NoneOAuth2 认证使用的 Bearer Token基本用法import daft from daft.catalog import Catalog catalog Catalog.from_gravitino( endpointhttp://localhost:8090, metalake_namemy_metalake, usernameadmin, ) # 列出某 schema 下所有表 tables catalog.list_tables(my_catalog.my_schema) # 读取表格式自动探测 df catalog.get_table(my_catalog.my_schema.my_table).read() df.show()表名采用catalog.schema.table三段式全限定名list_tables的 pattern 则支持catalog.schema两级、catalog一级以及无 pattern 时遍历整个 metalake 下的全部表——这一行为可以在 daft/catalog/__gravitino/_catalog.py 的_list_tables实现中验证。认证方式的底层实现Catalog.from_gravitino支持两种认证方法其真实落点在 GravitinoClient 的构造函数中Simple Authentication使用用户名/口令或仅用户名。源码中若提供password会设置 HTTP Basic Authsession.auth (username, password)若仅提供username则写入X-Gravitino-User请求头from daft.catalog import Catalog # Simple auth with username only catalog Catalog.from_gravitino( endpointhttp://localhost:8090, metalake_namemy_metalake, auth_typesimple, usernameadmin, )OAuth2使用 Bearer Token对应Authorization: Bearer token请求头# OAuth2 auth catalog Catalog.from_gravitino( endpointhttp://localhost:8090, metalake_namemy_metalake, auth_typeoauth2, tokenmy-bearer-token, )客户端的所有请求都通过requests.Session走{endpoint}/api/...路径的 REST 接口见 daft/catalog/__gravitino/_client.py 的_make_request这也是文档Limitations中所说直接调用 Gravitino RESTful API、未使用 Gravitino Python client的具体含义。存储凭据的自动提取Gravitino 通过表和 fileset 的 properties 管理存储凭据。Daft 客户端会自动提取并构造对应的IOConfig核心逻辑在 _io_config_from_storage_locationS3 / S3A从 properties 中读取 access key、secret key、session token 等且同时兼容两种键名风格——Fileset Catalog 的连字符格式s3-access-key-id、s3-secret-access-key与 Iceberg Catalog 的点分格式s3.access-key-id、s3.secret-access-key。如果未显式提供 region还会尝试从 endpoint URL如s3.ap-northeast-1.amazonaws.com中正则推断 region。Azureaz/abfs/abfss支持从azure.sas-token属性构造AzureConfig。GCS目前尚未完全支持会发出 warning 并返回None。其他 scheme会提示 Credentials for scheme ... are not yet supported 并返回None。值得注意的边界情况当 storage location 是本地路径且 Gravitino 返回file:/path时客户端会将其规范化为 Daft 期望的file:///path形式见 load_table。表读取格式自动探测与分发catalog.get_table(...)返回的表对象由 GravitinoTable._from_obj 按格式自动分发实现依次尝试 Iceberg、Hive/Parquet、Postgres 三类表根据 Gravitino 返回的format与table_type判定。Iceberg 表GravitinoIcebergTable 匹配format以ICEBERG开头的表读取时支持以下选项由Table._validate_options校验snapshot_id、branch、tag、ignore_corrupt_files。底层打开方式有两条路径见 _open_iceberg_tableREST 模式当表属性中存在catalog-backend: rest且配置了uri时通过 PyIceberg 的 REST catalog 加载并把 Daft 的IOConfig中 S3 凭据转写为 PyIceberg 的s3.*FileIO 属性若属性中没有凭据信息会显式声明auth{type: noop}避免 PyIceberg 发出Authorization: Bearer None导致 401Hadoop 模式否则使用HadoopCatalog以 storage location 的父目录为 warehouse 直接加载表目录。目前通过 Gravitino 对 Iceberg 表的写入append/overwrite尚未实现会抛出NotImplementedError。Hive/Parquet 表GravitinoParquetTable 匹配format PARQUET且table_type包含HIVE的表读取时直接调用read_parquet(pathstorage_location, hive_partitioningTrue)即自动启用 Hive 分区解析。写入同样尚未支持。Table.from_gravitinoTable.from_gravitino(table)可从已有的 GravitinoTable 对象创建 Daft Table定义见 daft/catalog/init.pyfrom daft.catalog import Catalog, Table catalog Catalog.from_gravitino( endpointhttp://localhost:8090, metalake_namemy_metalake, usernameadmin, ) table catalog.get_table(my_catalog.my_schema.my_table) df table.read()GVFS 协议通过 gvfs:// 读写 Fileset 文件Daft 支持通过gvfs://协议直接读取和写入 Gravitino fileset 中的文件为各种云存储S3、GCS 等提供统一访问接口屏蔽底层存储细节。GVFS URL 格式gvfs://fileset/catalog/schema/fileset/path其中catalogGravitino catalog 名称schemacatalog 内的 schema 名称filesetfileset 名称path可选fileset 内具体文件的路径。这一格式在 Rust IO 层由 GravitinoSource 严格解析scheme 必须为gvfshost 必须为fileset随后依次取前三段路径作为 catalog、schema、fileset剩余部分拼接为文件相对路径。解析失败时会给出形如 Expected Gravitino fileset path to be in the formgvfs://fileset/catalog/schema/fileset/path 的明确错误相关解析与错误处理有大量单元测试覆盖见 src/daft-io/src/gravitino.rs。配置 IOConfigGVFS 访问需要把 Gravitino 连接信息注入IOConfig。GravitinoConfig的字段定义见 daft/daft/init.pyiendpoint、metalake_name、auth_type、username、password、token其中password/token在 Rust 侧用ObfuscatedString存储以避免明文泄漏见 src/common/io-config/src/gravitino.rsimport daft from daft.io import IOConfig, GravitinoConfig # 构建用于 GVFS fileset 访问的 IOConfig io_config IOConfig( gravitinoGravitinoConfig( endpointhttp://localhost:8090, metalake_namemy_metalake, usernameadmin, ) )使用 GVFS 读取文件# 读取 fileset 中的 parquet 文件glob 通配 df daft.read_parquet( gvfs://fileset/my_catalog/my_schema/my_fileset/**/*.parquet, io_configio_config, ) # 读取指定文件 df daft.read_parquet( gvfs://fileset/my_catalog/my_schema/my_fileset/data.parquet, io_configio_config, ) # 读取 CSV 文件 df daft.read_csv( gvfs://fileset/my_catalog/my_schema/my_fileset/*.csv, io_configio_config, ) # 使用 glob 模式发现文件 files_df daft.from_glob_path( gvfs://fileset/my_catalog/my_schema/my_fileset/**/*.json, io_configio_config, )从 Rust 源码看glob 操作的实现细节比较有意思GravitinoSource::glob会先把gvfs://glob 模式映射为底层存储上的实际路径执行列举再把结果路径前缀从存储位置如s3://bucket/prefix/反向替换回gvfs://fileset/...前缀后返回——因此你在 DataFrame 中看到的文件路径始终保持为 gvfs 形式见 glob 实现。使用 GVFS 写入文件import daft from daft.io import IOConfig, GravitinoConfig io_config IOConfig( gravitinoGravitinoConfig( endpointhttp://localhost:8090, metalake_namemy_metalake, usernameadmin, ) ) # 构造样例数据 df daft.from_pydict({ id: [1, 2, 3], name: [Alice, Bob, Charlie], age: [25, 30, 35] }) # 写 parquet 到 fileset df.write_parquet( gvfs://fileset/my_catalog/my_schema/my_fileset/output.parquet, io_configio_config, ) # 写 CSV df.write_csv( gvfs://fileset/my_catalog/my_schema/my_fileset/output.csv, io_configio_config, ) # 写 JSON df.write_json( gvfs://fileset/my_catalog/my_schema/my_fileset/output.json, io_configio_config, )写入路径在 Python 侧由 GravitinoFileSystem 承接它包装了一个 FSSpec 风格的 handlerGravitinoFileSystemHandler让 PyArrow 的 parquet writer 等组件可以面向gvfs://URL 打开输出流GravitinoOutputStream 先缓冲数据在close()时通过 Rust 层的io_put一次性提交到 fileset 的实际存储位置。删除、移动、拷贝等操作目前未实现会抛NotImplementedError。GVFS 带来的收益统一访问读写使用同一套 URL 形式存储抽象无需感知底层是 S3、GCS 还是其他对象存储元数据集成借助 Gravitino 的 catalog 元数据进行数据发现凭据管理存储凭据由 Gravitino 侧集中处理客户端自动提取多格式支持兼容 Parquet、CSV、JSON 等文件格式。此外从源码结构看GravitinoSource内部维护了一个按catalog.schema.fileset名称缓存IOClient storage_location的并发 Mapget_or_create_io_client即同一 fileset 只向 Gravitino 元数据服务查询一次后续读写直接复用底层 IO 客户端避免重复的元数据往返。兼容性与限制API 版本兼容性该集成同时支持 Gravitino 的旧版与现行 API 格式见 load_tableLegacy 格式1.0 之前存储位置位于properties.location现行格式1.0多个存储位置位于storageLocations支持通过default-location-name属性缺省为default指定默认位置找不到时回退到第一个可用位置。客户端会自动检测并处理两种格式实现无缝兼容。当前限制Limitations尚未实现 credential vending凭据发放当前版本直接调用 Gravitino RESTful API未使用 Gravitino Python clientGVFS 写入目前仅适用于 S3 后端的 fileset其他存储后端支持即将提供部分高级 Gravitino 特性可能未通过该客户端暴露。路线图Roadmap支持从 Gravitino 读写 Iceberg 表 ✓已完成支持 Hive/Parquet 表 ✓已完成支持更多表格式Hudi支持更多存储GCS、Azure ADLS、OSS 等支持 credential vending集成测试如何在本地验证连接器仓库内置了一套针对 Gravitino 连接器的端到端集成测试位于 tests/integration/gravitino/可直接用于本地演练。按其 README 的步骤启动本地 Gravitino 服务cd tests/integration/gravitino/docker-compose docker compose up -d导出连接配置compose 文件的默认值与下列默认值一致export GRAVITINO_ENDPOINT${GRAVITINO_ENDPOINT:-http://127.0.0.1:8090} export GRAVITINO_METALAKE${GRAVITINO_METALAKE:-metalake_demo}运行测试DAFT_RUNNERnative pytest tests/integration/gravitino -v -m integration测试集包括test_gravitino_fileset.py本地file://存储的 fileset、test_gravitino_fileset_s3.pyS3 存储 fileset使用 MinIO、test_gravitino_table.py表目录操作。认证方式可通过环境变量控制见 conftest.pyGRAVITINO_AUTH_TYPEsimple/oauth2、GRAVITINO_USERNAME、GRAVITINO_PASSWORD、GRAVITINO_TOKENS3 场景还可将GRAVITINO_TEST_DIR/GRAVITINO_TEST_FILE指向你环境中真实存在的gvfs://fileset 路径。关键源码索引模块路径职责目录入口daft/catalog/init.pyCatalog.from_gravitino/Table.from_gravitino工厂方法REST 客户端daft/catalog/__gravitino/_client.py认证会话、catalog/schema/table/fileset 加载、凭据提取目录与表适配daft/catalog/__gravitino/_catalog.py格式自动分发、Iceberg/Hive 读取路径Rust IO 源src/daft-io/src/gravitino.rsgvfs://解析、fileset 客户端缓存、glob 路径变换配置文件结构src/common/io-config/src/gravitino.rsGravitinoConfig凭据脱敏存储PyArrow 文件桥接daft/io/gravitino_filesystem.pygvfs 输出流支撑write_parquet等写入集成测试tests/integration/gravitino/docker-compose 环境 端到端用例IO 层测试tests/io/test_gravitino_io.pyGVFS 读写行为验证如果你有当前连接器尚未覆盖的用法场景建议向 Daft 或 Gravitino 的开源仓库提交 issue 反馈推动 Roadmap 中 Hudi 格式、更多存储后端与 credential vending 能力的落地。【免费下载链接】DaftHigh-performance data engine for AI and multimodal workloads. Process images, audio, video, and structured data at any scale项目地址: https://gitcode.com/GitHub_Trending/da/Daft创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考