Hướng dẫn Webhook: Xử lý các tệp mới trong một bucket với Jobs

Storage Buckets là nơi phù hợp cho các tệp được tạo theo thời gian: dữ liệu xuất từ các công cụ khác, nhật ký, bản ghi âm, kết quả quét. Bạn thường sẽ có một quy trình cần thực hiện trên các tệp này trước khi chúng trở nên hữu ích, chẳng hạn như chuyển đổi, làm sạch hoặc chuyển thành văn bản.

Một webhook trên bucket và một Job có thể tự động thực hiện công việc này mà không cần máy chủ. Khi các tệp thay đổi, webhook sẽ khởi động Job. Job nhận danh sách các tệp đã thay đổi, xử lý chúng và dừng lại. Bạn chỉ phải trả tiền trong thời gian nó chạy.

Hướng dẫn này giải thích mô hình, sau đó thiết lập nó với một ví dụ: chuyển đổi các tệp CSV và JSON đã tải lên thành Parquet được tối ưu hóa.

Cách hoạt động

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

Webhook có thể khởi chạy một Job thay vì gọi một URL. Bạn tạo Job một lần, và mỗi sự kiện webhook sẽ chạy lại nó với sự kiện đó trong môi trường của nó. Đối với một bucket, WEBHOOK_PAYLOAD liệt kê các tệp đã thay đổi:

{
  "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 }
  ]
}

Do đó, Job chỉ xử lý các tệp mới và có thể quyết định cách xử lý từng tệp trước khi tải xuống bất kỳ tệp nào.

Mỗi lần chạy sẽ tái sử dụng image, lệnh, biến môi trường, cấu hình phần cứng và thời gian chờ của Job. Nó không nhận các volume hay secrets của Job. Đó là lý do các bước bên dưới chạy script từ một URL và đặt token vào webhook.

Job ghi kết quả của nó vào bucket thứ hai. Nếu nó ghi vào bucket mà nó đang theo dõi, chính đầu ra của nó sẽ kích hoạt webhook một lần nữa.

Ví dụ: chuyển các tệp tải lên thành Parquet đã tối ưu

Xuất CSV và JSON truy vấn chậm và khó chia sẻ. Parquet tối ưu lọc và stream nhanh hơn, đồng thời tải lên và tải xuống nhanh hơn nhờ khử trùng lặp Xet. Trong ví dụ này, mọi tệp CSV, JSON hoặc Parquet được tải lên một bucket sẽ xuất hiện dưới dạng Parquet tối ưu trong một bucket khác. Job chạy optimize-parquet.py, một script có sẵn từ uv-scripts.

Bạn cần một tài khoản Hugging Face có tiền trả trước và hf CLI.

Tạo hai bucket

Một bucket nhận các tệp bạn tải lên, và bucket còn lại nhận các tệp Parquet:

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

Tạo Job

Tạo Job mà webhook sẽ chạy:

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

Script được chạy từ URL của nó bằng hf jobs run. hf jobs uv run là cách thông thường để chạy một script UV, nhưng nó sẽ tải script lên một volume, mà một lần chạy webhook sẽ không có. cpu-upgrade là đủ cho công việc này, và thời gian chờ hai giờ để lại không gian cho các tệp lớn.

Script cũng có thể nằm trong một bucket hoặc repo riêng tư. Không sử dụng bucket được theo dõi: việc tải script lên sẽ khởi động một Job. Thêm --secrets HF_TOKEN vào lệnh, với một token có thể đọc script. Nếu run thất bại với SyntaxError, token không thể đọc được nó. Token của webhook cần cùng quyền truy cập, vì các run webhook không nhận được secret này.

hf jobs run khởi động Job ngay lập tức. Lần chạy đầu tiên này không có sự kiện webhook, và optimize-parquet.py không làm gì nếu không có sự kiện đó, nên Job sẽ dừng ngay. Sao chép Job ID mà nó in ra.

Tạo webhook

Xem bucket đầu tiên và chạy Job của bạn khi nó thay đổi:

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

--domain repo giới hạn webhook cho các thay đổi tệp và cài đặt. Buckets chỉ gửi những thay đổi đó, nên đây là sự rõ ràng thay vì bắt buộc, nhưng nó giữ cho lệnh đúng nếu sau này bạn theo dõi một model hoặc dataset.

--secrets HF_TOKEN lưu trữ một token với webhook, được mã hóa. Mọi Job do webhook khởi động đều nhận token này như một secret và sử dụng nó để đọc và ghi vào các bucket. Giá trị này được lấy từ HF_TOKEN trong môi trường của bạn, hoặc từ token bạn đã dùng để đăng nhập. Một fine-grained token chỉ chứa các quyền mà Job cần là lựa chọn an toàn nhất.

Để sử dụng token được tạo riêng cho pipeline này, hãy truyền giá trị một cách tường minh bằng --secrets HF_TOKEN=hf_…, hoặc giữ nó khỏi lịch sử shell bằng cách pipe vào: printf 'HF_TOKEN=hf_…\n' | hf webhooks create … --secrets-file -.

Lệnh này in ra ID của webhook. Để dừng pipeline sau đó, hãy xóa webhook bằng hf webhooks delete <webhook ID>.

Tải lên một tệp

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

Khoảng một phút sau, một Job sẽ xuất hiện trên trang Jobs của bạn. Nếu không thấy, hãy mở tab Activity của webhook trong cài đặt webhook của bạn để xem thông tin chuyển giao, và đọc đầu ra của Job bằng hf jobs logs <job ID>. Khi Job hoàn tất, tệp Parquet sẽ nằm trong bucket thứ hai:

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

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

Để chuyển đổi nhiều tệp, hãy tải chúng lên bằng một lệnh hf buckets sync. Các tệp được tải lên cùng lúc thường đến dưới dạng một sự kiện, do đó một Job sẽ chuyển đổi tất cả chúng, tối đa 10.000 tệp mỗi sự kiện. Xem Webhooks để biết chi tiết về payload của bucket và những gì xảy ra khi vượt quá giới hạn đó.

Những gì mà script thực hiện

Phần cụ thể cho webhook là đọc sự kiện:

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)

Với mỗi tệp mới, script sẽ nạp tệp đó bằng datasets trực tiếp từ bucket (hf://buckets/...). Nếu tệp lớn hơn một phần ba dung lượng đĩa trống của Job, nó sẽ được stream thay vì tải xuống. Sau đó, push_to_hub sẽ ghi tệp vào bucket đầu ra dưới dạng Parquet với chunking theo nội dung, chỉ mục trang và các nhóm hàng có kích thước tối đa 100 MB.

Sử dụng mẫu này cho tác vụ của riêng bạn

Giữ nguyên cấu hình và thay đổi script. Ví dụ:

• Xóa dữ liệu cá nhân trước khi huấn luyện. Thu thập văn bản thô trong một bucket riêng tư, che giấu thông tin cá nhân bằng một mô hình như GLiNER2 PII filter, và chỉ đẩy văn bản đã che giấu vào bộ dữ liệu huấn luyện của bạn.
• Đánh giá các checkpoint mới. Một Job huấn luyện lưu các checkpoint vào một bucket, và mỗi checkpoint mới sẽ khởi động một Job đánh giá. Nếu một checkpoint được tải lên theo nhiều phần, chỉ phản hồi khi tệp marker mà bạn ghi cuối cùng xuất hiện trong updatedFiles; các sự kiện có thể đến không theo thứ tự.
• Chuyển văn bản hoặc nhúng các tệp mới. Chuyển các bản ghi âm mới thành văn bản, hoặc các tài liệu mới thành các vector nhúng để tìm kiếm.

Khi viết script, hãy quyết định nó sẽ làm gì trong lần chạy đầu tiên, khi WEBHOOK_PAYLOAD chưa được đặt. Nó có thể thoát, như optimize-parquet.py làm, hoặc thực hiện công việc thực sự, chẳng hạn như xử lý các tệp đã có trong bucket. Nếu lần chạy đầu tiên cần token, ví dụ để đọc bucket hoặc một script riêng tư, hãy thêm --secrets HF_TOKEN vào hf jobs run. Các lần chạy webhook không nhận secret này. Chúng nhận token từ webhook.

Webhook hay lịch trình?

Webhook khởi chạy một Job cho mỗi sự kiện, do đó các tệp mới sẽ được xử lý khoảng một phút sau khi đến. Nếu không cần xử lý nhanh đến vậy, một Job được lên lịch là một lựa chọn thay thế: nó chạy mỗi giờ hoặc mỗi ngày và xử lý tất cả những gì đã đến kể từ lần chạy trước. Cách này gom công việc vào ít Job hơn, giúp ích cho các tác vụ GPU khởi động chậm và tránh giới hạn 1.000 kích hoạt mỗi 24 giờ của webhook. Khi đó, script phải tự tìm các tệp mới, ví dụ bằng cách so sánh các bucket đầu vào và đầu ra.