Руководство по Webhook: обработка новых файлов в бакете с помощью Jobs

Storage Buckets — удобное место для файлов, которые появляются со временем: выгрузки из других инструментов, журналы, записи, сканы. Часто требуется выполнить над этими файлами определённую последовательность действий, прежде чем они станут полезными, например, конвертацию, очистку или расшифровку.

Вебхук на бакете и Job могут выполнять эту работу автоматически, без сервера. Когда файлы изменяются, вебхук запускает Job. Job получает список изменённых файлов, обрабатывает их и останавливается. Вы платите только во время его работы.

В этом руководстве объясняется данный паттерн, а затем он настраивается на одном примере: преобразование загруженных файлов CSV и JSON в оптимизированный Parquet.

Как это работает

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

Вебхук может запускать Job вместо вызова URL. Вы создаёте 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 записывает свои результаты в второе ведро. Если бы он записывал в то ведро, за которым наблюдает, его собственный вывод снова бы сработал вебхук.

Пример: преобразование загрузок в оптимизированный 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

Создать Job

Создайте 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-скрипта, но он загружает скрипт на том, которого не будет при запуске через вебхук. cpu-upgrade достаточно для этой задачи, а двухчасовой таймаут оставляет запас для работы с большими файлами.

Скрипт также может находиться в приватном бакете или репозитории. Не используйте наблюдаемый бакет: загрузка скрипта запустит Job. Добавьте --secrets HF_TOKEN в команду, указав токен, который имеет доступ на чтение скрипта. Если запуск завершится ошибкой SyntaxError, это означает, что токен не может прочитать скрипт. Токен вебхука должен иметь тот же доступ, так как при запусках через вебхук этот секрет не передаётся.

hf jobs run запускает Job сразу. У этого первого запуска нет события webhook, а optimize-parquet.py без него ничего не делает, поэтому Job останавливается сразу. Скопируйте ID Job, который он выводит.

Создать вебхук

Следите за первым bucket и запускайте свой Job при его изменении:

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

--domain repo ограничивает вебхук изменениями файлов и настроек. Buckets отправляют только эти события, поэтому параметр не обязателен, но он делает команду корректной, если вы позже начнёте отслеживать модель или набор данных.

--secrets HF_TOKEN сохраняет токен вместе с вебхуком в зашифрованном виде. Каждый Job, запускаемый вебхуком, получает его как секрет и использует для чтения и записи в бакеты. Значение берётся из переменной окружения HF_TOKEN или из токена, с которым вы вошли в систему. Наиболее безопасный вариант — токен с тонкой настройкой, содержащий только те разрешения, которые нужны Job.

Чтобы использовать токен, созданный специально для этого конвейера, явно передайте его значение с помощью --secrets HF_TOKEN=hf_… или передайте его через конвейер, чтобы он не сохранялся в истории командной строки: printf 'HF_TOKEN=hf_…\n' | hf webhooks create … --secrets-file -.

Команда выводит ID вебхука. Чтобы остановить конвейер позже, удалите вебхук с помощью hf webhooks delete <webhook ID>.

Загрузить файл

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

Спустя примерно минуту на вашей странице Jobs появится Job. Если этого не произошло, откройте вкладку Activity вебхука в настройках вебхука, чтобы увидеть доставку, и прочитайте вывод Job с помощью hf jobs logs <job ID>. Когда 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 файлов на событие. См. Webhooks для полного описания полезной нагрузки бакета и того, что происходит при превышении этого лимита.

Что делает скрипт

Часть, специфичная для вебхуков, — это чтение события:

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 МБ.

Используйте шаблон для своей задачи

Сохраните конфигурацию и измените скрипт. Например:

• Удаляйте персональные данные перед обучением. Собирайте исходный текст в приватном бакете, удаляйте персональную информацию с помощью модели, например GLiNER2 PII filter, и отправляйте в набор данных для обучения только очищенный текст.
• Оценивайте новые контрольные точки. Job обучения сохраняет контрольные точки в бакет, и каждая новая контрольная точка запускает Job оценки. Если контрольная точка загружается несколькими частями, реагируйте только тогда, когда в updatedFiles появится маркерный файл, который вы записываете последним; события могут приходить в произвольном порядке.
• Транскрибируйте или создавайте эмбеддинги для новых файлов. Преобразуйте новые записи в транскрипты, а новые документы — в эмбеддинги для поиска.

При написании скрипта определите, что он делает при первом запуске, когда WEBHOOK_PAYLOAD не задан. Он может завершиться, как это делает optimize-parquet.py, или выполнить реальную работу, например, обработать файлы, которые уже находились в бакете. Если для первого запуска требуется токен, например, для чтения бакета или приватного скрипта, добавьте --secrets HF_TOKEN к hf jobs run. Запуски по вебхуку не получают этот секрет. Они получают свой токен из вебхука.

Вебхук или расписание?

Вебхук запускает Job для каждого события, поэтому новые файлы обрабатываются примерно через минуту после их поступления. Если такая скорость обработки не требуется, альтернативой может стать запланированный Job: он выполняется каждый час или каждый день и обрабатывает всё, что поступило с момента последнего запуска. Это позволяет сгруппировать работу в меньшее количество Jobs, что полезно для GPU-задач с длительным временем запуска, и позволяет обойти ограничение вебхука в 1000 срабатываний за 24 часа. В этом случае скрипт должен самостоятельно находить новые файлы, например, сравнивая входное и выходное хранилища.