数据工程数据湖大数据对象存储后端【免费下载链接】lakeFSlakeFS - Data version control for your data lake | Git for data项目地址https://gitcode.com/gh_mirrors/la/lakeFS点击查看免费下载导读ObjectsApi是 lakeFS Rust 客户端lakefs_sdk中负责对象Object级数据操作的 API 集合覆盖对象的上传、下载、复制、删除、元数据查询与修改等完整生命周期。本文以clients/rust/docs/ObjectsApi.md为核心骨架结合clients/rust/src/apis/objects_api.rs的实现代码与服务端pkg/api/controller.go的对应逻辑逐一剖析全部 10 个端点的函数签名、参数语义、认证与返回类型并给出可直接运行的 Rust 调用示例。读完本文你将能够用 Rust 实现对 lakeFS 分支对象的全链路读写与版本管理操作。一、API 概览方法与端点速查ObjectsApi中所有 URI 均以/api/v1为基准base path。文档中列出的全部端点如下表MethodRust 函数HTTP 请求说明copy_objectPOST /repositories/{repository}/branches/{branch}/objects/copy创建对象的副本delete_objectDELETE /repositories/{repository}/branches/{branch}/objects删除对象对象缺失时不会返回 NotFound 错误delete_objectsPOST /repositories/{repository}/branches/{branch}/objects/delete批量删除对象缺失对象不返回 NotFound 错误get_objectGET /repositories/{repository}/refs/{ref}/objects获取对象内容get_underlying_propertiesGET /repositories/{repository}/refs/{ref}/objects/underlyingProperties获取对象在底层存储上的属性head_objectHEAD /repositories/{repository}/refs/{ref}/objects检查对象是否存在list_objectsGET /repositories/{repository}/refs/{ref}/objects/ls列出指定前缀下的对象stat_objectGET /repositories/{repository}/refs/{ref}/objects/stat获取对象元数据update_object_user_metadataPUT /repositories/{repository}/branches/{branch}/objects/stat/user_metadata重写全部对象元数据upload_objectPOST /repositories/{repository}/branches/{branch}/objects上传对象所有端点均支持三种认证方式basic_authHTTP Basic、cookie_auth会话 Cookie与jwt_tokenBearer Token。从 Rust 实现看认证实际通过Configuration中的basic_auth与bearer_access_token字段注入请求头见 clients/rust/src/apis/configuration.rs。在函数签名中repository、branch、ref、path等路径/查询参数在实现中都会经过crate::apis::urlencode进行 URL 编码后拼接到请求行因此仓库名、分支名中包含特殊字符时无需手动转义见 clients/rust/src/apis/objects_api.rs。一个重要的语义区分ref与branch只读端点get_object、head_object、list_objects、stat_object、get_underlying_properties的路径参数是ref文档明确说明“a reference (could be either a branch or a commit ID)”即既可以传分支名也可以传提交 IDCommit ID实现对历史版本的只读访问写端点copy_object、delete_object、delete_objects、upload_object、update_object_user_metadata的路径参数是branch因为写入必须发生在可变的分支之上。二、copy_object创建对象的副本pub async fn copy_object( configuration: configuration::Configuration, repository: str, branch: str, // 目标分支 dest_path: str, // 目标路径相对分支 object_copy_creation: models::ObjectCopyCreation, // 复制参数请求体 ) - Resultmodels::ObjectStats, ErrorCopyObjectErrorHTTP 请求POST /repositories/{repository}/branches/{branch}/objects/copydest_path以查询参数传递请求体为 JSON。Content-Typeapplication/jsonAcceptapplication/json。返回类型models::ObjectStats即复制后新对象的统计信息。ObjectCopyCreation 请求体结构ObjectCopyCreation见 clients/rust/src/models/object_copy_creation.rs包含以下字段字段类型是否必填说明src_pathString必填源对象路径相对 refsrc_refOptionString可选源引用为空时默认使用目标分支作为源 refforceOptionbool可选是否强制覆盖目标路径上已存在的对象shallowOptionbool可选创建浅拷贝不复制实际数据。目前仅支持同一仓库、同一分支内的浅拷贝浅拷贝对象可能与垃圾回收GC和分支保留策略产生冲突需谨慎使用构造请求体示例let copy_creation models::ObjectCopyCreation { src_path: data/raw.parquet.to_string(), src_ref: Some(main.to_string()), force: Some(true), shallow: Some(false), }; let stats objects_api::copy_object(config, my-repo, feature-1, data/copied.parquet, copy_creation).await?; println!(copied to: {}, stats.path);服务端行为印证在服务端 pkg/api/controller.go 中CopyObject做了以下校验与处理权限校验调用方必须同时拥有源路径的fs:ReadObject权限和目标路径的fs:WriteObject权限对应 pkg/permissions/actions.go 中定义的ReadObjectAction与WriteObjectAction分支存在性检查目标branch不存在时返回404 branch not found默认源 refsrc_ref为空时回退为branch本身与 SDK 文档描述一致最终调用Catalog.CopyEntry完成复制并将源对象的 checksum、size、content_type、物理地址等打包成ObjectStats返回。需要说明从 SDK 生成的代码看copy_object的请求体是整体 JSON 序列化发送的且src_path为必填而shallow、force等为可选字段序列化时会通过skip_serializing_if Option::is_none自动省略空值。三、delete_object 与 delete_objects删除对象3.1 单个删除 delete_objectpub async fn delete_object( configuration: configuration::Configuration, repository: str, branch: str, path: str, // 相对分支的路径 force: Optionbool, // 默认 false no_tombstone: Optionbool, // 默认 false实验性 ) - Result(), ErrorDeleteObjectErrorHTTP 请求DELETE /repositories/{repository}/branches/{branch}/objectspath、force、no_tombstone均为查询参数。返回类型空响应体成功时返回Ok(())。文档特别注明对象缺失时不会返回 NotFound 错误删除是幂等的。3.2 批量删除 delete_objectspub async fn delete_objects( configuration: configuration::Configuration, repository: str, branch: str, path_list: models::PathList, // 请求体待删除路径列表 force: Optionbool, no_tombstone: Optionbool, ) - Resultmodels::ObjectErrorList, ErrorDeleteObjectsErrorHTTP 请求POST /repositories/{repository}/branches/{branch}/objects/delete请求体为 JSON查询参数为force、no_tombstone。Content-Typeapplication/jsonAcceptapplication/json。返回类型models::ObjectErrorList即逐个路径的操作错误列表。参数说明与请求体构造force类型Optionbool默认false。从服务端实现看它控制删除时的强制程度例如绕过对象的不可删除保护传入Some(true)时才会在查询串中携带forcetrue见 objects_api.rs。no_tombstone类型Optionbool默认false文档标注EXPERIMENTAL语义为“在可能的情况下无 tombstone 删除条目”。PathList见 clients/rust/src/models/path_list.rs只有一个字段paths: VecStringlet path_list models::PathList::new(vec![ logs/2024/01.log.to_string(), logs/2024/02.log.to_string(), ]); let errors objects_api::delete_objects(config, my-repo, main, path_list, Some(false), None).await?; // ObjectErrorList.errors: VecObjectErrorObjectError 结构批量删除的返回类型ObjectErrorList由VecObjectError组成见 clients/rust/src/models/object_error.rs字段类型说明status_codei32该路径操作对应的 HTTP 状态码messageString状态码的简短说明pathOptionString受影响的路径四、get_object下载对象内容pub async fn get_object( configuration: configuration::Configuration, repository: str, r#ref: str, // 分支名或提交 ID path: str, // 相对 ref 的路径 range: Optionstr, // 要获取的字节范围Range 头 if_none_match: Optionstr, // 仅当对象 ETag 不匹配时返回响应If-None-Match 头 presign: Optionbool, // 是否返回预签名 URL ) - Resultstd::path::PathBuf, ErrorGetObjectErrorHTTP 请求GET /repositories/{repository}/refs/{ref}/objectspath、presign为查询参数range通过Range请求头传递if_none_match通过If-None-Match请求头传递见 objects_api.rs。Acceptapplication/octet-stream, application/json。返回类型std::path::PathBuf。使用说明presign当传入Some(true)时响应返回的是一个预签名 HTTP URL而非内容本身从ObjectStats模型的定义可知预签名 URL 具有过期时间且可能因认证 token 即将过期而被截短见 clients/rust/src/models/object_stats.rs。SDK 层此模式下返回的PathBuf代表下载后保存到本地文件系统的路径。range支持 HTTP Range 语义例如Some(bytes0-1023)用于断点续传或只读部分数据。if_none_match用于条件请求例如传入对象当前 ETag可避免重复下载未变更的内容。五、head_object检查对象是否存在pub async fn head_object( configuration: configuration::Configuration, repository: str, r#ref: str, path: str, range: Optionstr, ) - Result(), ErrorHeadObjectErrorHTTP 请求HEAD /repositories/{repository}/refs/{ref}/objectspath为查询参数range为Range请求头。返回类型空响应体Ok(())表示对象存在Err(Error::ResponseError(...))表示不存在或请求失败。该端点是get_object的轻量版只校验对象的存在性与元数据不传输内容适合作为批处理任务前的存在性检查。从 SDK 生成的错误枚举看HeadObjectError覆盖 400/401/404/410/416/429 等状态码objects_api.rs。六、get_underlying_properties查询底层存储属性pub async fn get_underlying_properties( configuration: configuration::Configuration, repository: str, r#ref: str, path: str, ) - Resultmodels::UnderlyingObjectProperties, ErrorGetUnderlyingPropertiesErrorHTTP 请求GET /repositories/{repository}/refs/{ref}/objects/underlyingPropertiespath为查询参数。返回类型models::UnderlyingObjectProperties。Acceptapplication/json。该端点用于查询对象在底层对象存储S3、GS、Azure Blob 等上的原生属性。UnderlyingObjectProperties见 clients/rust/src/models/underlying_object_properties.rs当前只有一个字段字段类型说明storage_classOptionOptionString底层存储上的存储类别双层 Option 用于区分“未返回”与“显式 null”由serde_with::rust::double_option处理注意该端点返回的是底层存储的属性与stat_object返回的 lakeFS 元数据ObjectStats不同。七、list_objects列出前缀下的对象pub async fn list_objects( configuration: configuration::Configuration, repository: str, r#ref: str, user_metadata: Optionbool, // 默认 true presign: Optionbool, after: Optionstr, // 返回此值之后的条目 amount: Optioni32, // 返回条目数量默认 100 delimiter: Optionstr, // 用于按公共前缀分组的分隔符 prefix: Optionstr, // 只返回以该值为前缀的条目 ) - Resultmodels::ObjectStatsList, ErrorListObjectsErrorHTTP 请求GET /repositories/{repository}/refs/{ref}/objects/ls全部以查询参数传递。返回类型models::ObjectStatsList。Acceptapplication/json。参数语义详解参数默认值作用user_metadatatrue结果中是否包含用户自定义元数据ObjectStats.metadatapresign无结果中的physical_address是否返回预签名 HTTP URLafter无分页游标返回该值之后的条目配合Pagination.next_offset使用amount100每页返回的最大条目数delimiter无类似 S3 ListObjects 的分隔符如/将公共前缀折叠为path_type common_prefix的条目prefix无前缀过滤只列出以该前缀开头的对象返回结构 ObjectStatsListObjectStatsList见 clients/rust/src/models/object_stats_list.rs包含pagination: Pagination与results: VecObjectStats。Pagination见 clients/rust/src/models/pagination.rs提供字段类型说明has_morebool是否还有下一页next_offsetString用于获取下一页的游标值resultsi32本页结果数量max_per_pagei32每页最大条目数分页遍历示例let mut after: OptionString None; loop { let page objects_api::list_objects( config, my-repo, main, Some(true), None, after.as_deref(), Some(100), Some(/), Some(data/), ).await?; for obj in page.results { println!({} ({:?}), obj.path, obj.path_type); } if !page.pagination.has_more { break; } after Some(page.pagination.next_offset.clone()); }关于 path_type 与 ObjectStatsObjectStats见 clients/rust/src/models/object_stats.rs是对象操作中最核心的返回模型字段包括path、path_typecommon_prefix/object、physical_address底层存储原生 URI如s3://...、gs://...presign 时则为 HTTP URL、physical_address_expiry预签名 URL 过期时间Unix 秒、checksum、size_bytes、mtime、metadata用户自定义元数据 Map、content_type。其中size_bytes、metadata、content_type在返回时通常都会填充但在客户端上传等场景下允许缺省。八、stat_object获取对象元数据pub async fn stat_object( configuration: configuration::Configuration, repository: str, r#ref: str, path: str, user_metadata: Optionbool, // 默认 true presign: Optionbool, ) - Resultmodels::ObjectStats, ErrorStatObjectErrorHTTP 请求GET /repositories/{repository}/refs/{ref}/objects/statpath、user_metadata、presign为查询参数。返回类型models::ObjectStats结构详见上文。Acceptapplication/json。典型用途读取对象大小、checksum、mtime用于增量同步判断携带presigntrue获取可直接下载的预签名 URL 并交给其他工具消费结合head_object实现“先 stat 再决定是否下载”的缓存策略。九、update_object_user_metadata重写对象元数据pub async fn update_object_user_metadata( configuration: configuration::Configuration, repository: str, branch: str, // 要更新的分支 path: str, // 相对分支的对象路径 update_object_user_metadata: models::UpdateObjectUserMetadata, // 请求体 ) - Result(), ErrorUpdateObjectUserMetadataErrorHTTP 请求PUT /repositories/{repository}/branches/{branch}/objects/stat/user_metadatapath为查询参数请求体为 JSON。Content-Typeapplication/jsonAcceptapplication/json。返回类型空响应体。语义为rewrite (all) object metadata即整体重写对象的用户元数据而非增量合并。UpdateObjectUserMetadata见 clients/rust/src/models/update_object_user_metadata.rs只有一个字段set: HashMapString, String表示要设置的完整元数据键值集合let mut metadata std::collections::HashMap::new(); metadata.insert(owner.to_string(), data-team.to_string()); metadata.insert(stage.to_string(), prod.to_string()); let req models::UpdateObjectUserMetadata::new(metadata); objects_api::update_object_user_metadata(config, my-repo, main, data/raw.parquet, req).await?;由于是“整体重写”调用前应先用stat_object获取现有元数据再在内存中合并修改后整体提交避免丢失已有键值。十、upload_object上传对象pub async fn upload_object( configuration: configuration::Configuration, repository: str, branch: str, path: str, // 相对分支的路径 if_none_match: Optionstr, // *仅在路径尚不存在时允许上传 if_match: Optionstr, // 对象的 ETag仅当当前 ETag 匹配时允许操作 storage_class: Optionstr, // 已弃用未来版本不再支持 force: Optionbool, // 默认 false content: Optionstd::path::PathBuf, // 单文件上传必须命名为 content ) - Resultmodels::ObjectStats, ErrorUploadObjectErrorHTTP 请求POST /repositories/{repository}/branches/{branch}/objectspath、storageClass、force为查询参数if_none_match通过If-None-Match请求头、if_match通过If-Match请求头传递。Content-Typemultipart/form-data或application/octet-streamAcceptapplication/json。返回类型models::ObjectStats上传完成后对象的统计信息。参数语义与注意事项参数说明if_none_match设为*可实现原子性“仅当该路径尚无对象时才允许上传”即 create-if-absent其他值不支持。服务端在 pkg/api/controller.go 中对该值做了专门处理若路径已存在则返回412 Precondition Failed并将条件封装为 graveler 的WithCondition参与写入if_match设为对象当前 ETag可实现 compare-and-set 式原子更新防止并发覆盖storage_class文档明确标注Deprecated未来版本将移除该能力force默认false用于控制是否强制覆盖已有对象content单文件上传multipart 表单中的文件字段必须命名为content服务端双通道实现服务端UploadObjectpkg/api/controller.go根据Content-Type走两条路径application/octet-stream非 multipart直接把请求体作为对象内容写入底层存储upload.WriteBlob适合流式上传multipart/form-data解析 multipart 中的content字段逐段读取并上传。同时服务端会先校验调用方对目标路径是否拥有fs:WriteObject权限、仓库与分支是否存在分支不存在直接返回404 branch not found避免在错误分支上白白上传大文件。注意当前 SDK 生成的upload_object对content参数只创建了空 multipart form代码中留有// TODO: support file upload for content parameter见 objects_api.rs。因此在当前生成版本中若需要携带文件内容上传可改用application/octet-stream方式直传自行构造请求或关注该 TODO 的后续实现。十一、错误处理与认证配置类型化错误枚举objects_api.rs为每个端点定义了对应的类型化错误枚举如CopyObjectError、UploadObjectError、GetObjectError等每个枚举通过#[serde(untagged)]反序列化服务端返回的 JSON 错误体并覆盖该端点可能的 HTTP 状态码。例如GetObjectError400 / 401 / 404 / 410 / 416 / 429410 Gone 与 416 Range Not Satisfiable 常见于对象过期与非法 RangeUploadObjectError400 / 401 / 403 / 404 / 412 / 429 / 501412 对应If-None-Match: *冲突501 对应未支持的能力如已弃用的 storage class 相关操作所有端点均包含429 Too Many Requests限流与DefaultResponse未预期错误。调用时通过ErrorXxxError的ResponseError变体即可拿到status、content与反序列化后的错误实体match objects_api::stat_object(config, my-repo, main, nope.txt, Some(true), None).await { Ok(stats) println!({} bytes, stats.size_bytes.unwrap_or(0)), Err(e) match e { apis::Error::ResponseError(resp) { eprintln!(HTTP {}: {}, resp.status, resp.content); } _ eprintln!(transport error: {:?}, e), }, }配置与认证所有端点共享同一个Configurationclients/rust/src/apis/configuration.rs默认base_path为/api/v1。在真实部署中应将其替换为完整的 lakeFS 服务地址例如http://localhost:8000/api/v1并配置认证let config apis::configuration::Configuration { base_path: http://localhost:8000/api/v1.to_owned(), basic_auth: Some(( AKIAIOSFODNN7EXAMPLE.to_owned(), Some(wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY.to_owned()), )), ..Default::default() };lakeFS 的访问密钥对Access Key / Secret Key即对应 HTTP Basic 认证的用户名/密码若使用 Bearer Token如jwt_token则设置bearer_access_token字段。每次调用传入config即可reqwest::Client在内部复用连接。十二、对象操作的最佳实践小结综合文档与源码用ObjectsApi构建对象工作流时有几点值得注意读写分离的 ref 语义查询/下载一律走ref分支名或 Commit ID实现“读历史版本”写入走branch保证数据只落在可变的当前分支上。原子写保护需要“不存在才创建”时用upload_object(if_none_matchSome(*))需要“防并发覆盖”时用if_matchSome(etag)配合stat_object先取 ETag。分页与前缀list_objects用prefix过滤 after/Pagination.next_offset游标翻页delimiter可模拟目录层级common_prefix。预签名访问get_object、stat_object、list_objects都支持presigntrue可将底层对象存储的访问以临时 URL 形式安全地交给第三方消费。幂等删除delete_object/delete_objects对缺失对象不报错适合在清理任务中直接使用批量删除返回ObjectErrorList可针对每个路径的错误单独处理。元数据整体重写update_object_user_metadata是“全量替换”语义修改前先stat_object拉取现值避免误丢键值。以上即为ObjectsApi的完整端点图谱与实战要点。进一步可参考 SDK 目录下的模型文档如 ObjectStats、ObjectCopyCreation以及服务端实现 pkg/api/controller.go以便在需要时深入定制行为。赞分享数据工程数据湖大数据对象存储后端【免费下载链接】lakeFSlakeFS - Data version control for your data lake | Git for data项目地址https://gitcode.com/gh_mirrors/la/lakeFS点击查看免费下载相关推荐lakeFS Java SDK ObjectsApi 完全指南对象读写、复制、删除与元数据操作实战lakeFS Java SDK ObjectsApi 完全指南对象读写、复制、删除与元数据操作实战 本指南以 lakeFS 官方 Java SDK io.l数据工程数据湖大数据对象存储后端lakeFS Python SDK 数据对象操作完全指南ObjectsApi 十个 API 的实战用法与源码解析lakeFS Python SDK 数据对象操作完全指南ObjectsApi 十个 API 的实战用法与源码解析 本文以 lakeFS 官方 Python S数据工程数据湖大数据对象存储后端lakeFS Rust SDK 对象复制ObjectCopyCreation 模型详解与 copy_object 实战lakeFS Rust SDK 对象复制ObjectCopyCreation 模型详解与 copy_object 实战 导读 本文围绕 lakeFS Rust数据工程数据湖大数据对象存储后端上一篇如何快速掌握Maxun工作流引擎Where-What条件配对机制详解下一篇sam2_hiera_base_plus.fb_r896_2pt1与Transformers集成指南轻松实现图像特征提取创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考