数据工程数据湖大数据对象存储后端【免费下载链接】lakeFSlakeFS - Data version control for your data lake | Git for data项目地址https://gitcode.com/gh_mirrors/la/lakeFS点击查看免费下载本文围绕 lakeFS Java SDK 参考文档中的TaskCreation模型展开它是一个仅有id一个必填属性的极简响应模型承载着 lakeFS 异步垃圾回收GC准备任务的任务标识。读者读完本文后将掌握TaskCreation的 OpenAPI 定义、它在prepare gc commits异步端点中的唯一出场位置、从 HTTP 202 响应到服务端后台任务落库的完整调用链以及如何在 Java / Python SDK 中消费任务 ID 并轮询任务状态。从一份极简模型文档说起TaskCreation.md 是 lakeFS Java SDKio.lakefs.clients.sdk中由 OpenAPI Generator 自动生成的模型参考文档其正文只有一张属性表NameTypeDescriptionNotesidStringThe id of the new task看似简单但这张表背后对应的是 lakeFS HTTP API 中一类典型的异步任务句柄设计服务端接到耗时操作请求后不立即返回最终结果而是先创建一个后台任务把任务标识id交还给客户端客户端再凭此id轮询任务状态。TaskCreation就是这个任务句柄的载体在本仓库中它专用于异步 GC 提交准备prepare garbage collection commits流程。TaskCreation 的 OpenAPI 定义唯一属性 id 及其约束在 API 规范 swagger.yml 中TaskCreation的 schema 定义如下TaskCreation: type: object properties: id: description: The id of the new task type: string required: - id要点如下type: object它是一个标准的 JSON 对象响应模型。idstring必填新任务new task的标识符由服务端生成并返回客户端后续所有针对该任务的操作都依赖它。required: [id]意味着服务端保证响应体中一定携带id字段。该模型没有其他属性也不接受 OpenAPI 层面的额外扩展字段但在 Java 客户端模型中出于兼容性考虑仍保留了额外未声明属性additional properties的透传机制详见下文模型类实现细节。Java SDK 中对应的模型类是 TaskCreation.java其static初始化块将id同时注册进openapiFields与openapiRequiredFields必填字段集合并在validateJsonElement中强制校验响应 JSON 中必须存在id且id必须是 JSON 原始类型primitive否则直接抛出IllegalArgumentException// 校验必填字段是否缺失 for (String requiredField : TaskCreation.openapiRequiredFields) { if (jsonElement.getAsJsonObject().get(requiredField) null) { throw new IllegalArgumentException(...); } } if (!jsonObj.get(id).isJsonPrimitive()) { throw new IllegalArgumentException( String.format(Expected the field id to be a primitive type in the JSON string but got %s, ...)); }TaskCreation 的唯一出场位置异步 GC 准备端点在整个 API 规范中TaskCreation只被一处引用——内部internal接口组的异步 GC 准备端点。相关定义见 swagger.yml/repositories/{repository}/gc/prepare_commits/async: parameters: - in: path name: repository required: true schema: type: string post: tags: - internal operationId: prepareGarbageCollectionCommitsAsync summary: prepare gc commits responses: 202: description: GC prepare task started content: application/json: schema: $ref: #/components/schemas/TaskCreation也就是说客户端调用POST /api/v1/repositories/{repository}/gc/prepare_commits/async服务端以HTTP 202 Accepted返回TaskCreation表示GC 准备任务已启动这是你的任务 ID。值得注意的是这里的设计取舍同步版本prepareGarbageCollectionCommitsPOST /repositories/{repository}/gc/prepare_commits在 swagger.yml 中直接返回201 Created和GarbageCollectionPrepareResponse含run_id、gc_commits_location、gc_addresses_location及可选的预签名地址适合数据量小、可短时完成的场景。异步版本本文主题返回 202 TaskCreation将真正耗时的扫描并导出活跃提交列表下沉为后台任务适合大型仓库避免请求长时间挂起。端点身份与返回类型在 InternalApi.java 中对应的生成方法签名为public APIprepareGarbageCollectionCommitsAsyncRequest prepareGarbageCollectionCommitsAsync(String repository)服务端路由注册可参见生成的 lakefs.gen.goPOST /repositories/{repository}/gc/prepare_commits/async被分发到ServerInterfaceWrapper.PrepareGarbageCollectionCommitsAsync。服务端全链路从 HTTP 202 到任务落库Controller 层权限校验与响应组装真正的服务端实现位于 controller.gofunc (c *Controller) PrepareGarbageCollectionCommitsAsync(w http.ResponseWriter, r *http.Request, repository string) { if !c.authorize(w, r, permissions.Node{ Permission: permissions.Permission{ Action: permissions.PrepareGarbageCollectionCommitsAction, Resource: permissions.RepoArn(repository), }, }) { return } ctx : r.Context() c.LogAction(ctx, prepare_garbage_collection_commits_async, r, repository, , ) taskID, err : c.Catalog.PrepareExpiredCommitsAsync(ctx, repository) if c.handleAPIError(ctx, w, r, err) { return } writeResponse(w, r, http.StatusAccepted, apigen.TaskCreation{ Id: taskID, }) }执行流程可归纳为三步权限校验调用者必须持有PrepareGarbageCollectionCommitsAction权限且作用于该仓库RepoArn(repository)否则请求在入口即被拒绝。创建后台任务调用 Catalog 层的PrepareExpiredCommitsAsync拿到任务 ID。组装响应把taskID填入apigen.TaskCreation{Id: taskID}以 202 状态码写回客户端——这就是TaskCreation.id的最终来源。Catalog 层任务步骤的注册Catalog 层实现见 catalog.go。PrepareExpiredCommitsAsync首先校验仓库 ID 合法性graveler.ValidateRepositoryID并检查仓库只读状态随后构造一个任务步骤TaskStep该步骤真正的工作是调用c.Store.SaveGarbageCollectionCommits(ctx, repository)生成 GC 运行元数据run id、commits CSV 位置、地址位置并将结果写入任务状态对象taskStatus : GarbageCollectionPrepareStatus{} taskSteps : []TaskStep{ { Name: prepare expired commits on repository.RepositoryID.String(), Func: func(ctx context.Context) error { gcRunMetadata, err : c.Store.SaveGarbageCollectionCommits(ctx, repository) if err ! nil { return err } taskStatus.Info GarbageCollectionPrepareCommitsInfo{ ... } return nil }, }, } taskID : NewTaskID(GarbageCollectionPrepareCommitsPrefix) err c.RunBackgroundTaskSteps(ctx, repository, OpGCPrepareCommits, taskID, taskSteps, taskStatus)注意RunBackgroundTaskSteps本身在 catalog.go 中定义它会立即返回后台执行任务步骤这正是202 快速响应得以成立的机制任务执行期间及执行完毕后状态都会持续写入底层存储供状态查询接口读取。任务 ID 的生成规则GCPC 前缀 20 位 nanoid任务 ID 的格式可以直接从代码确认。前缀常量定义在 catalog.goGarbageCollectionPrepareCommitsPrefix GCPC生成函数在 task.gofunc NewTaskID(prefix string) string { return prefix nanoid.Must(taskIDNanoLength) } func IsTaskID(prefix, taskID string) bool { return len(taskID) len(prefix)taskIDNanoLength strings.HasPrefix(taskID, prefix) }其中taskIDNanoLength 20见 task.go。因此TaskCreation.id的实际形态为GCPC 20 位 nanoid共 24 字符例如GCPCxxxxxxxxxxxxxxxxxxxx。IsTaskID正是用前缀匹配 固定长度来校验 ID 归属防止跨任务类型串读状态。任务状态的 KV 存储与心跳机制任务状态并非内存态而是持久化到 lakeFS 的 KV 存储中。核心存取函数同样在 task.gofunc UpdateTaskStatus(ctx context.Context, kvStore kv.Store, repository *graveler.RepositoryRecord, taskID string, statusMsg protoreflect.ProtoMessage) error { return kv.SetMsg(ctx, kvStore, graveler.RepoPartition(repository), []byte(TaskPath(taskID)), statusMsg) } func GetTaskStatus(ctx context.Context, kvStore kv.Store, repository *graveler.RepositoryRecord, taskID string, statusMsg protoreflect.ProtoMessage) (kv.Predicate, error) { predicate, err : kv.GetMsg(ctx, kvStore, graveler.RepoPartition(repository), []byte(TaskPath(taskID)), statusMsg) ... }任务消息以仓库分区graveler.RepoPartition下的tasks/{taskID}路径保存天然具备跨实例可见性。此外从 catalog.go 的markTaskExpiredIfStale可以看出系统还引入了**实例心跳instance heartbeat**机制带OwnerInstanceId的任务通过检查属主实例在 KV 中的心跳记录判断任务是否僵死若属主实例失联超过过期时长任务会被标记为Done并写入 Task owner instance no longer available. Please retry. 之类的错误信息从而避免客户端无限等待。状态查询端点用 id 换任务进度有了id客户端通过配套端点查询进度见 swagger.ymlGET /api/v1/repositories/{repository}/gc/prepare_commits/status?id{task_id}返回模型PrepareGarbageCollectionCommitsStatusswagger.yml字段如下字段类型说明task_idstring正在准备 GC commits 的任务 ID即TaskCreation.idcompletedboolean任务是否已结束无论成功或出错update_timedate-time任务状态最后更新时间resultGarbageCollectionPrepareResponse任务成功后的结果含run_id、gc_commits_location、gc_addresses_location等errorError任务失败时的错误信息其中task_id、completed、update_time三个字段为必填。服务端实现在 controller.go取出查询参数id调用c.Catalog.GetGarbageCollectionPrepareStatuscatalog.go后者经GetValidatedTaskStatus校验任务 ID 前缀并读取 KV 中的任务状态最终把TaskId、Completed、UpdateTime以及成功时的result/ 失败时的error组装进响应。Java SDK 实战如何消费 TaskCreation完整调用示例Java SDK 参考文档 与 InternalApi.java 给出了完整的调用方式。以下示例基于仓库文档整理可直接运行// Import classes: import io.lakefs.clients.sdk.ApiClient; import io.lakefs.clients.sdk.ApiException; import io.lakefs.clients.sdk.Configuration; import io.lakefs.clients.sdk.auth.*; import io.lakefs.clients.sdk.model.*; import io.lakefs.clients.sdk.InternalApi; public class Example { public static void main(String[] args) { ApiClient defaultClient Configuration.getDefaultApiClient(); defaultClient.setBasePath(http://lakefs-host/api/v1); // Configure HTTP basic authorization: basic_auth HttpBasicAuth basic_auth (HttpBasicAuth) defaultClient.getAuthentication(basic_auth); basic_auth.setUsername(YOUR USERNAME); basic_auth.setPassword(YOUR PASSWORD); // 可选cookie_authAPI key // ApiKeyAuth cookie_auth (ApiKeyAuth) defaultClient.getAuthentication(cookie_auth); // cookie_auth.setApiKey(YOUR API KEY); // 可选jwt_tokenHTTP bearer // HttpBearerAuth jwt_token (HttpBearerAuth) defaultClient.getAuthentication(jwt_token); // jwt_token.setBearerToken(BEARER TOKEN); InternalApi apiInstance new InternalApi(defaultClient); String repository repository_example; // 仓库名必填 try { // 返回的正是 TaskCreation只有 id 一个字段 TaskCreation result apiInstance.prepareGarbageCollectionCommitsAsync(repository) .execute(); System.out.println(task id: result.getId()); } catch (ApiException e) { System.err.println(Exception when calling InternalApi#prepareGarbageCollectionCommitsAsync); System.err.println(Status code: e.getCode()); System.err.println(Reason: e.getResponseBody()); System.err.println(Response headers: e.getResponseHeaders()); e.printStackTrace(); } } }该端点支持三种认证方式basic_auth、cookie_auth、jwt_token请求无请求体Content-Type 未定义响应格式为application/json。三种执行模式生成的请求类APIprepareGarbageCollectionCommitsAsyncRequest提供三种执行模式InternalApi.java// 1) 同步阻塞直接返回 TaskCreation TaskCreation result request.execute(); // 2) 同步 HTTP 元信息状态码、响应头 ApiResponseTaskCreation resp request.executeWithHttpInfo(); // 3) 异步回调 okhttp3.Call call request.executeAsync(new ApiCallbackTaskCreation() { Override public void onFailure(ApiException e, int statusCode, MapString, ListString responseHeaders) { } Override public void onSuccess(TaskCreation result, int statusCode, MapString, ListString responseHeaders) { } Override public void onUploadProgress(long bytesWritten, long contentLength, boolean done) { } Override public void onDownloadProgress(long bytesRead, long contentLength, boolean done) { } });拿到TaskCreation.getId()后即可调用prepareGarbageCollectionCommitsStatus(repository, id).execute()轮询状态直到completed true再取result中的 GC 输出位置执行后续清理。TaskCreation 在 Java 模型类中的实现细节TaskCreation.java 由 OpenAPI Generator 基于swagger.yml自动生成几个值得留意的实现点字段与序列化SERIALIZED_NAME_ID id配合SerializedName(id)绑定 JSON 键名提供链式id(String id)、getId()、setId()三个入口。Gson 支持内置CustomTypeAdapterFactory序列化时移除内部additionalProperties占位、再把未声明属性逐个写入 JSON反序列化时先validateJsonElement再填充实例未知字段会落入additionalProperties映射保证向后兼容。便捷转换fromJson(String)/toJson()基于全局JSON.getGson()便于调试与测试。对象语义equals/hashCode以id与 additional properties 为判等依据toString输出人类可读的缩进形式。对应的单元测试骨架见 TaskCreationTest.javaAPI 层测试见 InternalApiTest.java。其他语言 SDK 中的对应实现TaskCreation不只是 Java SDK 的专属模型其他语言客户端同样生成了等价类型Python SDK模型类 task_creation.py 用 pydantic 定义class TaskCreation(BaseModel): id: StrictStr Field(..., descriptionThe id of the new task)其中id为必填StrictStr并提供了to_json/from_json/to_dict/from_dict等序列化辅助方法。API 方法位于 internal_api.py同样支持同步调用与async_reqTrue异步线程两种模式from lakefs_sdk.api.internal_api import InternalApi from lakefs_sdk import ApiClient, Configuration configuration Configuration(hosthttp://lakefs-host/api/v1) configuration.username YOUR USERNAME configuration.password YOUR PASSWORD api_client ApiClient(configuration) api InternalApi(api_client) # 同步返回 TaskCreationid 字段即任务 ID task api.prepare_garbage_collection_commits_async(repositorymy-repo) print(task.id) # 异步线程方式 thread api.prepare_garbage_collection_commits_async(repositorymy-repo, async_reqTrue) result thread.get()Go 服务端生成代码pkg/api/apigen/lakefs.gen.go中同时生成了请求构造器、WithResponse封装与响应解析函数如 ParsePrepareGarbageCollectionCommitsAsyncResponse供服务端与 Go 客户端复用同一套类型契约。常见响应状态与排错prepareGarbageCollectionCommitsAsync的响应状态码完整清单来源swagger.yml 与 InternalApi.md状态码说明常见触发场景202GC prepare task started返回TaskCreation正常响应400Validation Error仓库名非法如违反graveler.ValidateRepositoryID规则401Unauthorized未提供/提供错误的基本认证或令牌403Forbidden已认证但缺少PrepareGarbageCollectionCommitsAction权限404Resource Not Found指定仓库不存在429too many requests触发限流0 / 5xxInternal Server Error服务端异常排错建议收到 403 时检查当前凭证是否被授予针对该仓库的 GC 相关权限。收到 404 时先确认仓库名拼写及repository路径参数是否与 API 一致。拿到TaskCreation.id后若查询状态返回 404可核对任务 ID 是否完整应满足GCPC前缀 20 位 nanoid 的长度与格式因为IsTaskID会对前缀和长度做严格校验。状态查询中completed true且携带error字段时说明后台任务已失败可结合error信息排查存储层对象存储、KV 库的可达性与凭证配置。小结TaskCreation是 lakeFS 为异步 GC 准备流程设计的极简任务句柄模型服务端在POST /repositories/{repository}/gc/prepare_commits/async上以 202 快速返回{id: GCPCxxxx…}后台通过RunBackgroundTaskSteps执行提交扫描与导出任务状态持久化于 KV 存储客户端凭id通过/gc/prepare_commits/status轮询PrepareGarbageCollectionCommitsStatus。理解这一模型就能正确地在 Java、Python、Go 等多语言 SDK 中串联提交 GC 任务 → 获取任务 ID → 轮询状态 → 消费 GC 结果的完整闭环。延伸阅读均位于当前仓库TaskCreation.md —— 本文主体文档Java SDK 模型参考InternalApi.md ——prepareGarbageCollectionCommitsAsync完整 Java 示例TaskCreation.java —— Java 模型实现swagger.yml ——TaskCreationschema 与端点定义controller.go —— 服务端 Controller 实现catalog.go —— Catalog 层后台任务注册task.go —— 任务 ID 生成与 KV 存取task_creation.py —— Python SDK 对应模型赞分享数据工程数据湖大数据对象存储后端【免费下载链接】lakeFSlakeFS - Data version control for your data lake | Git for data项目地址https://gitcode.com/gh_mirrors/la/lakeFS点击查看免费下载相关推荐lakeFS Rust SDK 中 TaskCreation 模型详解异步 GC 任务句柄与完整使用链路lakeFS Rust SDK 中 TaskCreation 模型详解异步 GC 任务句柄与完整使用链路 TaskCreation 是 lakeFS Rust数据工程数据湖大数据对象存储后端GitHub MCP Server 在 Copilot IDE 中的安装配置实战Visual Studio、JetBrains、Xcode 与 Eclipse 完整指南GitHub MCP Server 在 Copilot IDE 中的安装配置实战Visual Studio、JetBrains、Xcode 与 Eclipse数据工程数据湖大数据对象存储后端lakeFS Rust SDK 深度解析GarbageCollectionPrepareResponse 模型与 GC Commits 准备流程lakeFS Rust SDK 深度解析GarbageCollectionPrepareResponse 模型与 GC Commits 准备流程 导读 Gar数据工程数据湖大数据对象存储后端创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考