• 数据工程
• 数据湖
• 大数据
• 对象存储
• 后端

导读

StagingApi 是 lakeFS Python SDK（lakefs_sdk）中负责"暂存（staging）"写入路径的核心 API：它通过get_physical_address与link_physical_address两个方法，把"上传数据对象到对象存储"和"把对象登记到 lakeFS 仓库"两个动作解耦，从而在不经过 lakeFS 服务器中转数据的情况下安全写入对象。阅读完本文，你将掌握 lakeFS 暂存写入的完整工作流、StagingLocation/StagingMetadata/ObjectStats三个关键模型的字段语义、presign与 ETag 条件写入的实战用法，以及服务端与 SDK 中的对应实现细节。

一、StagingApi 概览：两个方法完成一次暂存写入

StagingApi 的全部端点都挂在/api/v1之下，共两个方法，对应两条 HTTP 请求路径：

方法签名定义在 clients/python/lakefs_sdk/api/staging_api.py 中：

def get_physical_address(self, repository, branch, path, presign=None, **kwargs) -> StagingLocation def link_physical_address(self, repository, branch, path, staging_metadata, if_none_match=None, if_match=None, **kwargs) -> ObjectStats

调用示例、参数表与返回类型详见 clients/python/docs/StagingApi.md；服务端的行为契约定义在 api/swagger.yml 的/repositories/{repository}/branches/{branch}/staging/backing路径下（operationId 分别为getPhysicalAddress与linkPhysicalAddress）。

1.1 两种认证方式

两个方法都支持三种认证，SDK 生成的客户端在 staging_api.py 中固定声明_auth_settings = ['basic_auth', 'cookie_auth', 'jwt_token']：

• basic_auth：HTTP Basic 认证，配置username与password；
• cookie_auth：API Key（Cookie），通过configuration.api_key['cookie_auth']设置，需要前缀时可用configuration.api_key_prefix['cookie_auth'] = 'Bearer'；
• jwt_token：Bearer JWT，通过Configuration(access_token=...)配置。

1.2 为什么需要"暂存"两步

lakeFS 的对象存储（blockstore）直接面向 S3、GCS、Azure Blob 等后端。两步设计的好处是：

1. 数据不经过 lakeFS 服务器：客户端直接向对象存储上传数据（或直接引用外部已存在的对象地址），避免服务器吞吐成为瓶颈；
2. 元数据与数据分离：上传完成后再通过link_physical_address把对象的 checksum、大小、路径等元数据登记进仓库，形成未提交变更，随后可用Commit固化；
3. 支持外部地址：link_physical_address允许关联"仓库存储命名空间之外"的地址，为数据导入、跨仓库引用等场景提供可能。

二、get_physical_address：生成上传地址

2.1 请求参数

2.2 返回模型 StagingLocation

返回类型为StagingLocation，定义在 clients/python/lakefs_sdk/models/staging_location.py，共三个字段：

2.3 完整调用示例

import time import os import lakefs_sdk from lakefs_sdk.models.staging_location import StagingLocation from lakefs_sdk.rest import ApiException from pprint import pprint # 配置认证：Basic configuration = lakefs_sdk.Configuration( username=os.environ["USERNAME"], password=os.environ["PASSWORD"] ) # 或配置 JWT # configuration = lakefs_sdk.Configuration(access_token=os.environ["BEARER_TOKEN"]) with lakefs_sdk.ApiClient(configuration) as api_client: api_instance = lakefs_sdk.StagingApi(api_client) repository = 'repo_name' branch = 'main' path = 'data/part-0001.parquet' # 相对分支的路径 presign = True api_response = api_instance.get_physical_address(repository, branch, path, presign=presign) pprint(api_response)

HTTP 请求头：Accept: application/json，无请求体。状态码：200（成功，返回暂存物理地址）、400、401、404、429，其余为500内部错误。

2.4 服务端实现

在 pkg/api/controller.go 的GetPhysicalAddress处理器中，服务端完成以下工作：

1. 鉴权：要求WriteObjectAction权限（controller.go）；
2. 通过Catalog.GetAddressWithSignature为对象生成带签名的物理地址；
3. 通过BlockAdapter.ResolveNamespace把相对地址解析为完整对象存储 URI；
4. 若presign=true，调用BlockAdapter.GetPreSignedURL(..., block.PreSignModeWrite, "")生成写入型预签名 URL，并填充PresignedUrl与PresignedUrlExpiry字段（controller.go）。

三、link_physical_address：把物理地址与路径关联

3.1 请求参数

if_none_match与if_match在 SDK 生成代码中分别映射为 HTTP 头If-None-Match与If-Match（见 staging_api.py），二者提供原子化的写入条件，用于避免并发覆盖或重复写入。

3.2 请求体模型 StagingMetadata

StagingMetadata定义在 clients/python/lakefs_sdk/models/staging_metadata.py，字段如下：

3.3 返回模型 ObjectStats

link_physical_address成功返回ObjectStats（200 状态码），定义在 clients/python/lakefs_sdk/models/object_stats.py。关键字段：

3.4 完整调用示例

import os import lakefs_sdk from lakefs_sdk.models.object_stats import ObjectStats from lakefs_sdk.models.staging_metadata import StagingMetadata from lakefs_sdk.rest import ApiException from pprint import pprint configuration = lakefs_sdk.Configuration( username=os.environ["USERNAME"], password=os.environ["PASSWORD"] ) with lakefs_sdk.ApiClient(configuration) as api_client: api_instance = lakefs_sdk.StagingApi(api_client) repository = 'repo_name' branch = 'main' path = 'data/part-0001.parquet' # 第一步：取得上传地址（假定此前已完成上传并拿到 ETag） staging_location = lakefs_sdk.StagingLocation( physical_address="s3://.../...", presigned_url="https://...", ) staging_metadata = lakefs_sdk.StagingMetadata( staging=staging_location, checksum="2e9ec317e197e02e4264d128c2e7e681", # 后端返回的 ETag size_bytes=1048576, user_metadata={"team": "data-eng"}, content_type="application/octet-stream", mtime=None, force=False, ) if_none_match = '*' # 仅当路径尚无对象时允许写入 if_match = None # 或传入对象当前 ETag 做条件写入 api_response = api_instance.link_physical_address( repository, branch, path, staging_metadata, if_none_match=if_none_match, if_match=if_match, ) pprint(api_response)

HTTP 请求头：Content-Type: application/json，Accept: application/json。状态码语义：

3.5 服务端实现

controller.go 的LinkPhysicalAddress处理器揭示了完整流程：

1. 鉴权要求WriteObjectAction（controller.go）；
2. 校验请求体中的 blockstore 类型与仓库存储配置一致；
3. 若addressType == catalog.AddressTypeRelative（地址位于仓库存储命名空间内），调用Catalog.VerifyLinkAddress校验该地址确由 lakeFS 签发（controller.go）；命名空间之外的地址则直接接受；
4. 使用httputil.StripQuotesAndSpaces清洗 ETag，若为空则返回 400（controller.go）；
5. 通过catalog.NewDBEntryBuilder()构造对象条目（路径、物理地址、大小、checksum、ContentType、元数据），并用apiutil.BuildConditionFromParams将if_match/if_none_match转换为 graveler 条件（controller.go）；
6. Catalog.CreateEntry(..., graveler.WithForce(...), graveler.WithCondition(...))写入未提交条目（controller.go）。

这里的CreateEntry走的是 graveler 层的 staging 管理：pkg/graveler/staging/目录下的实现（如 pkg/graveler/staging/manager_test.go）即负责未提交（staged）数据的读写，最终这些暂存条目会在 Commit 时被固化。

四、完整实战：预签名两段式上传

结合presign=true与两个 StagingApi 方法，可以实现不经过 lakeFS 服务器的直传流程（参考官方端到端测试 esti/presign_test.go 中的preSignGetPhysicalAddress用例）：

1. 获取预签名上传地址：GET .../staging/backing?path=foo/uploaded&presign=true，得到StagingLocation（含presigned_url）；
2. 直传数据：向presigned_url发起 HTTPPUT，携带对象内容与Content-Type，从响应头Etag取得校验和；
3. 链接对象：PUT .../staging/backing，请求体StagingMetadata中staging回传第一步的结果、checksum填 ETag、size_bytes填实际字节数；
4. 读取验证：GET Object拉回对象内容，确认与上传内容一致。

Python 代码骨架：

import requests # 1. 获取预签名地址 loc = api_instance.get_physical_address(repo, branch, "foo/uploaded", presign=True) # 2. 直传对象到预签名 URL resp = requests.put(loc.presigned_url, data=obj_content, headers={"Content-Type": "application/octet-stream"}) etag = resp.headers.get("Etag") # 3. 链接物理地址与路径 stats = api_instance.link_physical_address( repo, branch, "foo/uploaded", staging_metadata=lakefs_sdk.StagingMetadata( staging=loc, checksum=etag, size_bytes=len(obj_content), content_type="application/octet-stream", ), )

4.1 高阶客户端中的实际用法

在更上层的 Python 封装clients/python-wrapper中，预签名上传正是依赖 StagingApi 完成的。见 clients/python-wrapper/lakefs/object.py 的_upload_presign实现：

staging_location = self._client.sdk_client.staging_api.get_physical_address( self._obj.repo, self._obj.ref, self._obj.path, True) url = staging_location.presigned_url # ... 直接向 url 发起 PUT，提取 ETag ... staging_metadata = StagingMetadata(staging=staging_location, size_bytes=size_bytes, checksum=etag, user_metadata=self.metadata, content_type=self.content_type, mtime=None) if_none_match = "*" if self._mode.startswith("x") else None return self._client.sdk_client.staging_api.link_physical_address( self._obj.repo, self._obj.ref, self._obj.path, staging_metadata=staging_metadata, if_none_match=if_none_match)

注意两个细节：写入模式为x（排他创建）时自动设置if_none_match="*"，服务器返回 412 时映射为ObjectExistsException；Azure blockstore 场景下还会额外添加x-ms-blob-type: BlockBlob头（object.py）。对应的单测见 clients/python-wrapper/tests/utests/test_object.py。

Go 侧参考实现位于 pkg/api/helpers/upload.go 的uploadObject：它同样先调用getPhysicalAddress（Presign=true），用http.MethodPut直传，再以LinkPhysicalAddressJSONRequestBody关联元数据；若响应为 409（提交冲突），返回ErrConflict供上层重试。

五、条件写入与冲突处理最佳实践

• 排他创建（不存在才写）：传if_none_match='*'，服务端仅在路径无对象时写入，返回 412 表示对象已存在；这是实现"对象级原子创建"的推荐手段。
• 乐观并发更新（ETag 匹配才写）：传if_match=当前ETag，仅当对象 ETag 与给定值一致时允许覆盖，避免"读-改-写"之间的并发丢失更新。
• 提交冲突（409）：链接发生在某次提交进行时可能返回 409 并携带新的StagingLocation，客户端应重新执行"获取地址 → 上传 → 链接"流程（Go 客户端在 upload.go 中实现了带重试的上传循环）。
• 外部地址链接：link_physical_address接受仓库存储命名空间之外的物理地址，但命名空间内的地址会被校验签名，防止伪造 lakeFS 地址。

六、小结

StagingApi 是 lakeFS 写入路径的关键拼图：get_physical_address解决"往哪传"，link_physical_address解决"传完怎么登记"。配合presign与 ETag 条件头，可以在保证原子性与并发安全的前提下实现高吞吐的对象直传。若要进一步探索：

• API 契约：查看 api/swagger.yml 中staging标签下的getPhysicalAddress/linkPhysicalAddress定义；
• 服务端实现：阅读 pkg/api/controller.go；
• 端到端验证：运行或阅读 esti/presign_test.go；
• 上层封装：参考 clients/python-wrapper/lakefs/object.py 与 pkg/api/helpers/upload.go。

• 数据工程
• 数据湖
• 大数据
• 对象存储
• 后端