Guia de Webhooks: Processar novos arquivos em um bucket com Jobs

Storage Buckets são um bom lugar para arquivos que chegam ao longo do tempo: exportações de outras ferramentas, logs, gravações, varreduras. Você frequentemente terá um fluxo de trabalho que precisa ser realizado nesses arquivos antes que eles sejam úteis, como convertê-los, limpá-los ou transcrevê-los.

Um webhook no bucket e um Job podem executar esse trabalho automaticamente, sem um servidor. Quando os arquivos mudam, o webhook inicia o Job. O Job recebe a lista de arquivos alterados, processa-os e para. Você paga apenas enquanto ele está em execução.

Este guia explica o padrão e, em seguida, o configura com um exemplo: converter arquivos CSV e JSON enviados em Parquet otimizado.

Como funciona

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

Um webhook pode disparar um Job em vez de chamar uma URL. Você cria o Job uma vez e cada evento de webhook o executa novamente com o evento em seu ambiente. Para um bucket, WEBHOOK_PAYLOAD lista os arquivos que foram alterados:

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

Portanto, o Job processa apenas os novos arquivos e pode decidir como lidar com cada um antes de baixar qualquer coisa.

Cada execução reutiliza a imagem, o comando, as variáveis de ambiente, o sabor de hardware e o tempo limite do Job. Ele não recebe os volumes do Job nem seus segredos. É por isso que as etapas abaixo executam o script a partir de uma URL e colocam o token no webhook.

O Job grava seus resultados em um segundo bucket. Se ele gravasse no bucket que ele monitora, sua própria saída dispararia o webhook novamente.

Exemplo: transformar uploads em Parquet otimizado

Exportações em CSV e JSON são lentas para consultas e difíceis de compartilhar. Parquet otimizado é mais rápido para filtrar e transmitir, e faz upload e download mais rápido graças à deduplicação do Xet. Neste exemplo, cada arquivo CSV, JSON ou Parquet enviado para um bucket aparece como Parquet otimizado em outro. O Job executa optimize-parquet.py, um script pronto de uv-scripts.

Você precisa de uma conta do Hugging Face com créditos pré-pagos e o hf CLI.

Crie dois buckets

Um bucket recebe seus uploads e o outro recebe os arquivos Parquet:

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

Criar o Job

Crie o Job que o webhook executará:

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

O script é executado a partir da sua URL, com hf jobs run. hf jobs uv run é a forma usual de executar um script UV, mas ele faz o upload do script para um volume, o que uma execução via webhook não teria. cpu-upgrade é suficiente para esse trabalho, e o timeout de duas horas deixa espaço para arquivos grandes.

O script também pode estar em um bucket ou repositório privado. Não use o bucket monitorado: o upload do script iniciaria um Job. Adicione --secrets HF_TOKEN ao comando, com um token que possa ler o script. Se a execução falhar com um SyntaxError, o token não consegue lê-lo. O token do webhook precisa do mesmo acesso, pois as execuções via webhook não recebem esse segredo.

hf jobs run inicia o Job imediatamente. Esta primeira execução não possui evento de webhook, e optimize-parquet.py não faz nada sem um, então o Job para imediatamente. Copie o ID do Job que ele imprime.

Crie o webhook

Observe o primeiro bucket e execute seu Job quando ele for alterado:

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

--domain repo restringe o webhook a alterações em arquivos e configurações. Os buckets enviam apenas essas alterações, então isso é explícito em vez de necessário, mas mantém o comando correto caso você monitore um modelo ou dataset no futuro.

--secrets HF_TOKEN armazena um token no webhook, criptografado. Todo Job iniciado pelo webhook o recebe como um secret e o utiliza para ler e gravar nos buckets. O valor vem de HF_TOKEN no seu ambiente, ou do token com o qual você fez login. Um token de granularidade fina com apenas as permissões necessárias para o Job é a opção mais segura.

Para usar um token criado apenas para este pipeline, passe o valor explicitamente com --secrets HF_TOKEN=hf_…, ou mantenha-o fora do histórico do seu shell enviando-o por pipe: printf 'HF_TOKEN=hf_…\n' | hf webhooks create … --secrets-file -.

O comando exibe o ID do webhook. Para interromper o pipeline posteriormente, exclua o webhook com hf webhooks delete <webhook ID>.

Enviar um arquivo

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

Aproximadamente um minuto depois, um Job aparece na sua página de Jobs. Se nenhum aparecer, abra a aba Activity do webhook em suas configurações de webhook para ver a entrega e leia a saída de um Job com hf jobs logs <job ID>. Quando o Job terminar, o arquivo Parquet estará no segundo bucket:

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

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

Para converter muitos arquivos, envie-os com um único hf buckets sync. Arquivos enviados juntos geralmente chegam como um único evento, de modo que um único Job converte todos eles, até 10.000 arquivos por evento. Consulte Webhooks para o payload completo do bucket e o que acontece acima desse limite.

O que o script faz

A parte específica de webhooks é a leitura do evento:

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)

Para cada novo arquivo, o script o carrega com datasets diretamente do bucket (hf://buckets/...). Se o arquivo for maior que um terço do disco livre do Job, ele é transmitido em streaming em vez de ser baixado. O push_to_hub então o grava no bucket de saída como Parquet, com chunking definido pelo conteúdo, um índice de página e grupos de linhas de no máximo 100 MB.

Use o padrão para sua própria tarefa

Mantenha a configuração e altere o script. Por exemplo:

• Remova dados pessoais antes do treinamento. Colete o texto bruto em um bucket privado, remova informações pessoais com um modelo como GLiNER2 PII filter e envie apenas o texto redigido para o seu conjunto de dados de treinamento.
• Avalie novos checkpoints. Um Job de treinamento salva checkpoints em um bucket, e cada novo checkpoint inicia um Job de avaliação. Se um checkpoint for enviado em várias partes, reaja apenas quando um arquivo marcador que você escreve por último aparecer em updatedFiles; os eventos podem chegar fora de ordem.
• Transcreva ou incorpore novos arquivos. Transforme novas gravações em transcrições ou novos documentos em incorporações para busca.

Ao escrever seu script, decida o que ele faz na primeira execução, quando WEBHOOK_PAYLOAD não está definido. Ele pode encerrar, como optimize-parquet.py faz, ou realizar trabalho real, como processar os arquivos que já estavam no bucket. Se a primeira execução precisar de um token, por exemplo para ler o bucket ou um script privado, adicione --secrets HF_TOKEN ao hf jobs run. As execuções via webhook não recebem esse segredo. Elas obtêm seu token do webhook.

Webhook ou agendamento?

Uma webhook inicia um Job para cada evento, portanto, novos arquivos são processados cerca de um minuto após chegarem. Se não for necessário processá-los com tanta rapidez, um Job agendado é uma alternativa: ele é executado a cada hora ou a cada dia e processa tudo que chegou desde a última execução. Isso agrupa o trabalho em menos Jobs, o que ajuda em tarefas de GPU que demoram para iniciar, e evita o limite da webhook de 1.000 acionamentos a cada 24 horas. O script deve, então, encontrar os novos arquivos por conta própria, por exemplo, comparando os buckets de entrada e saída.