Webhook 指南：使用 Jobs 处理存储桶中的新文件

Storage Buckets 是存放随时间陆续到达的文件的理想位置：来自其他工具的导出文件、日志、录音、扫描结果等。这些文件在变得可用之前，通常需要经过一系列处理流程，例如转换、清洗或转录。

通过存储桶上的 webhook 和一个 Job，可以自动完成这项工作，无需服务器。当文件发生变化时，webhook 会启动该 Job。Job 会接收已更改文件的列表，对其进行处理，然后停止。您只需在 Job 运行期间付费。

本指南将介绍该模式，然后通过一个示例进行演示：将上传的 CSV 和 JSON 文件转换为优化后的 Parquet 格式。

工作原理

file uploaded ──▶ webhook ──▶ Job runs with the event ──▶ result written to a second bucket

Webhook 可以触发一个 Job，而不是调用 URL。你只需创建一次 Job，每个 webhook 事件都会再次运行它，并将事件信息放入其环境中。对于 bucket，WEBHOOK_PAYLOAD 会列出发生变化的文件：

{
  "event": { "action": "update", "scope": "repo.content" },
  "repo": { "type": "bucket", "name": "your-username/my-raw-files", ... },
  "updatedFiles": [
    { "path": "data.csv", "action": "add", "xetHash": "433b53b8...", "size": 10983 }
  ]
}

因此，Job 仅处理新文件，并且可以在下载任何内容之前决定如何处理每个文件。

每次运行都会复用该 Job 的镜像、命令、环境变量、硬件规格和超时设置。它不会获取该 Job 的卷或密钥。因此，以下步骤通过 URL 运行脚本，并将令牌放在 webhook 上。

该 Job 会将结果写入第二个存储桶。如果它写入自己监控的存储桶，其自身的输出将再次触发 webhook。

示例：将上传转换为优化后的 Parquet

CSV 和 JSON 导出文件的查询速度较慢，且不便共享。优化后的 Parquet 在过滤和流式处理方面速度更快，并且得益于 Xet 去重技术，上传和下载速度也更快。在此示例中，上传到某个存储桶的每个 CSV、JSON 或 Parquet 文件都会以优化后的 Parquet 格式出现在另一个存储桶中。该任务运行 optimize-parquet.py，这是来自 uv-scripts 的现成脚本。

您需要一个拥有预付费额度的 Hugging Face 账户，以及hf CLI。

创建两个存储桶

一个存储桶用于接收您的上传文件，另一个用于接收 Parquet 文件：

hf buckets create my-raw-files --private
hf buckets create my-parquet --private

创建任务

创建 Webhook 将运行的 Job：

hf jobs run --flavor cpu-upgrade --timeout 2h \
    -e OUTPUT_BUCKET=your-username/my-parquet \
    ghcr.io/astral-sh/uv:python3.12-bookworm \
    uv run https://huggingface.co/datasets/uv-scripts/data-processing/raw/main/optimize-parquet.py

该脚本通过其 URL 使用 hf jobs run 运行。hf jobs uv run 是运行 UV 脚本的常规方式，但它会将脚本上传到卷中，而 webhook 运行时不具备此条件。对于此任务，cpu-upgrade 已足够，且两小时的超时时间为处理大文件留出了余地。

脚本也可以位于私有存储桶或仓库中。请勿使用被监控的存储桶：上传脚本会触发一个 Job。在命令中添加 --secrets HF_TOKEN，并使用一个能够读取该脚本的 token。如果运行因 SyntaxError 而失败，则说明该 token 无法读取脚本。Webhook 的 token 也需要相同的访问权限，因为 webhook 运行不会获取此 secret。

hf jobs run 会立即启动该 Job。由于首次运行没有 webhook 事件，而 optimize-parquet.py 在没有事件时不会执行任何操作，因此 Job 会立即停止。请复制其打印出的 Job ID。

创建 webhook

监控第一个存储桶，并在其内容发生变化时运行你的 Job：

hf webhooks create --job-id <job ID> --watch bucket:your-username/my-raw-files \
    --domain repo --secrets HF_TOKEN

--domain repo 将 webhook 限制为文件和设置变更。Buckets 仅发送这些变更，因此该参数是显式声明而非必需，但如果您之后监控模型或数据集，该命令仍然正确。

--secrets HF_TOKEN 会将令牌加密后存储到 webhook 中。webhook 启动的每个 Job 都会将其作为 secret 接收，并用于读写存储桶。该值来自您环境变量中的 HF_TOKEN，或您登录时使用的令牌。仅包含 Job 所需权限的 细粒度令牌 是最安全的选择。

若要使用专为该流水线创建的令牌，请通过 --secrets HF_TOKEN=hf_… 显式传入其值，或者通过管道传入以避免将其保留在 shell 历史记录中：printf 'HF_TOKEN=hf_…\n' | hf webhooks create … --secrets-file -。

该命令会打印 webhook ID。如需稍后停止流水线，请使用 hf webhooks delete <webhook ID> 删除该 webhook。

上传文件

hf buckets cp data.csv hf://buckets/your-username/my-raw-files/data.csv

大约一分钟后，您的 Jobs 页面 上会出现一个 Job。如果没有出现，请打开 webhook 设置 中的 Activity 选项卡以查看投递情况，并使用 hf jobs logs <job ID> 读取 Job 的输出。当 Job 完成后，Parquet 文件将位于第二个存储桶中：

hf buckets ls your-username/my-parquet -R

data.csv/README.md
data.csv/data/train-00000-of-00001.parquet

要转换大量文件，请使用一次 hf buckets sync 命令上传。一起上传的文件通常会作为一个事件到达，因此一个 Job 可以转换它们全部，每个事件最多支持 10,000 个文件。有关完整的 bucket 负载以及超出该限制后的行为，请参阅 Webhooks。

该脚本的功能

与 Webhook 相关的特定部分是读取事件：

event = json.loads(os.environ.get("WEBHOOK_PAYLOAD", "{}"))  # empty when you run the Job yourself
input_bucket = os.environ.get("WEBHOOK_REPO_ID")

for changed_file in event.get("updatedFiles", []):
    if changed_file["action"] != "add":
        continue  # skip deleted files
    path, size = changed_file["path"], changed_file["size"]
    ...  # convert the file (see the full script)

对于每个新文件，脚本会通过 datasets 直接从存储桶（hf://buckets/...）加载该文件。如果文件大于 Job 可用磁盘空间的三分之一，则会以流式方式处理，而不是下载。push_to_hub 随后将其以 Parquet 格式写入输出存储桶，采用基于内容的分块、页面索引以及最大 100 MB 的行组。

将此模式应用于您自己的任务

保留配置，修改脚本。例如：

• 在训练前移除个人数据。在私有存储桶中收集原始文本，使用诸如 GLiNER2 PII filter 之类的模型对个人信息进行脱敏处理，然后仅将脱敏后的文本推送到您的训练数据集中。
• 评估新的检查点。 训练 Job 会将检查点保存到存储桶中，每个新的检查点都会触发一个评估 Job。如果检查点被分多个部分上传，请仅在您最后写入的标记文件出现在 updatedFiles 中时再做出响应；因为事件可能会乱序到达。
• 转录或嵌入新文件。 将新的录音转换为转录文本，或将新文档转换为用于搜索的嵌入向量。

编写脚本时，请确定在 WEBHOOK_PAYLOAD 未设置的首次运行中，脚本应执行什么操作。它可以像 optimize-parquet.py 那样直接退出，也可以执行实际工作，例如处理桶中已有的文件。如果首次运行需要令牌（例如用于读取桶或私有脚本），请在 hf jobs run 中添加 --secrets HF_TOKEN。Webhook 运行不会获取此密钥。它们从 Webhook 获取令牌。

Webhook 还是定时任务？

Webhook 会为每个事件启动一个 Job，因此新文件会在到达后约一分钟开始处理。如果不需要如此快速地处理，定时 Job 是一个替代方案：它每小时或每天运行一次，并处理自上次运行以来到达的所有内容。这将工作合并到更少的 Job 中，有助于启动缓慢的 GPU 任务，并避免了 Webhook 每 24 小时 1,000 次触发的限制。此时，脚本必须自行查找新文件，例如通过比较输入和输出存储桶。