Webhookガイド: Jobsでバケット内の新しいファイルを処理する

Storage Bucketsは、時間とともに蓄積されるファイルの格納に適した場所です。他のツールからのエクスポート、ログ、録音、スキャンなどが該当します。これらのファイルが有用になる前に、変換、クリーニング、文字起こしなどの処理を行うワークフローを構築する必要があることがよくあります。

バケットに設定されたWebhookとJobを使用すると、サーバーなしでこの作業を自動的に行うことができます。ファイルが変更されると、WebhookがJobを起動します。Jobは変更されたファイルのリストを受け取り、それらを処理して停止します。実行中の時間のみ課金されます。

このガイドでは、まずパターンを説明し、その後、アップロードされたCSVファイルとJSONファイルを最適化されたParquetに変換する1つの例を通じてセットアップを行います。

仕組み

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

WebhookはURLを呼び出す代わりに、Jobをトリガーできます。Jobは一度作成すればよく、各Webhookイベントは、そのイベントを環境変数として含めてJobを再実行します。バケットの場合、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 は結果を2番目のバケットに書き込みます。監視しているバケットに書き込むと、Job 自身の出力が Webhook を再度トリガーしてしまうためです。

例: アップロードを最適化されたParquetに変換する

CSVおよびJSONのエクスポートはクエリが遅く、共有しにくい形式です。最適化されたParquetはフィルタリングとストリーミングが高速で、Xetの重複排除によりアップロードとダウンロードも高速になります。この例では、1つのバケットにアップロードされたすべてのCSV、JSON、Parquetファイルが、別のバケットに最適化されたParquetとして表示されます。このジョブは、uv-scriptsから提供される完成済みのスクリプトであるoptimize-parquet.pyを実行します。

事前購入クレジットとhf CLIを備えた Hugging Face アカウントが必要です。

2つのバケットを作成します

バケットにはアップロードされたファイルが、もう一方のバケットには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

スクリプトはhf jobs runによってそのURLから実行されます。hf jobs uv runはUVスクリプトを実行する通常の方法ですが、スクリプトをボリュームにアップロードするため、Webhook実行では利用できません。この作業にはcpu-upgradeで十分であり、2時間のタイムアウトは大きなファイルにも対応する余裕があります。

スクリプトはプライベートなバケットまたはリポジトリに配置することもできます。監視対象のバケットは使用しないでください。スクリプトをアップロードするとジョブが開始されてしまうためです。スクリプトを読み取れるトークンを使用し、コマンドに--secrets HF_TOKENを追加してください。実行がSyntaxErrorで失敗した場合は、トークンがスクリプトを読み取れていないことを意味します。Webhookの実行ではこのシークレットが取得されないため、Webhookのトークンにも同じアクセス権限が必要です。

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 は、このトークンをシークレットとして受け取り、バケットの読み書きに使用します。値は、環境変数の HF_TOKEN から、またはログイン時に使用したトークンから取得されます。Job に必要な権限のみを持つ ファイングレイントークン を使用するのが最も安全です。

このパイプライン専用のトークンを使用するには、--secrets HF_TOKEN=hf_… で値を明示的に渡すか、シェル履歴に残さないためにパイプで入力します: 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

約1分後、JobsページにJobが表示されます。表示されない場合は、Webhook設定のWebhookのActivityタブを開いて配信状況を確認し、hf jobs logs <job ID>でJobの出力を読み取ってください。Jobが完了すると、Parquetファイルは2番目のバケットに配置されます。

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

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

多数のファイルをコンバートするには、hf buckets syncで一度にアップロードしてください。一緒にアップロードされたファイルは通常、1つのイベントとして届くため、1つのJobで最大10,000ファイルまですべてコンバートされます。バケットのペイロード全体と、この上限を超えた場合の動作については、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の空きディスクの3分の1を超える場合は、ダウンロードする代わりにストリーミングされます。その後、push_to_hubは、コンテンツ定義チャンクング、ページインデックス、および最大100 MBの行グループを備えたParquet形式で、出力バケットに書き込みます。

このパターンを独自のタスクに適用する

セットアップはそのままに、スクリプトを変更します。例えば:

• 学習前に個人データを削除します。 生テキストをプライベートなバケットに収集し、GLiNER2 PII filterなどのモデルで個人情報をマスキングし、マスキング済みのテキストのみを学習データセットにプッシュします。
• 新しいチェックポイントを評価します。 学習 Job はチェックポイントをバケットに保存し、新しいチェックポイントごとに評価 Job が開始されます。チェックポイントが複数の部分に分割してアップロードされる場合は、最後に書き込むマーカーファイルがupdatedFilesに出現したときにのみ反応してください。イベントは順序が乱れて到着する可能性があります。
• 新しいファイルを文字起こしまたは埋め込みに変換します。 新しい録音データを文字起こしに変換したり、新しいドキュメントを検索用の埋め込みに変換したりできます。

スクリプトを作成する際は、WEBHOOK_PAYLOAD が設定されていない最初の実行時に何を行うかを決定してください。optimize-parquet.py のように終了することも、バケットに既に存在するファイルの処理など実際の作業を実行することもできます。最初の実行でトークンが必要になる場合（バケットの読み取りやプライベートなスクリプトの読み取りなど）、hf jobs run に --secrets HF_TOKEN を追加してください。Webhook による実行では、このシークレットは提供されません。トークンは Webhook から取得されます。

Webhookですか、スケジュールですか？

Webhook は各イベントで Job を開始するため、新しいファイルは到着から約1分後に処理されます。それほど迅速な処理が不要な場合は、スケジュールされた Jobが代替手段として機能します。これは毎時または毎日実行され、前回の実行以降に到着したすべてのファイルを処理します。これにより作業がより少ない数の Job にまとめられ、起動に時間がかかる GPU タスクに有利になり、24時間あたり1,000トリガーという Webhook の制限を回避できます。その場合、スクリプトは新しいファイルを自分で特定する必要があります。例えば、入力バケットと出力バケットを比較する方法などが考えられます。