Guide des webhooks : Traiter de nouveaux fichiers dans un bucket avec Jobs

Storage Buckets constituent un bon emplacement pour les fichiers qui arrivent au fil du temps : des exports d'autres outils, des journaux, des enregistrements, des scans. Vous aurez souvent un flux de travail à effectuer sur ces fichiers avant qu'ils ne soient utiles, tel que la conversion, le nettoyage ou la transcription.

Un webhook sur le bucket et un Job peuvent effectuer ce travail automatiquement, sans serveur. Lorsque des fichiers changent, le webhook démarre le Job. Le Job reçoit la liste des fichiers modifiés, les traite puis s’arrête. Vous ne payez que pendant son exécution.

Ce guide explique le modèle, puis le met en place à travers un exemple : la conversion de fichiers CSV et JSON téléversés en Parquet optimisé.

Fonctionnement

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

Un webhook peut déclencher un Job au lieu d’appeler une URL. Vous créez le Job une seule fois, et chaque événement de webhook le relance avec l’événement dans son environnement. Pour un bucket, WEBHOOK_PAYLOAD répertorie les fichiers qui ont changé :

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

Ainsi, le Job ne traite que les nouveaux fichiers et peut décider comment gérer chacun d'entre eux avant de télécharger quoi que ce soit.

Chaque exécution réutilise l’image, la commande, les variables d’environnement, le type de matériel et le délai d’expiration du Job. Elle n’hérite pas des volumes ni des secrets du Job. C’est pourquoi les étapes ci-dessous exécutent le script depuis une URL et transmettent le jeton via le webhook.

Le Job écrit ses résultats dans un deuxième bucket. S'il écrivait dans le bucket qu'il surveille, sa propre sortie déclencherait à nouveau le webhook.

Exemple : convertir les téléversements en Parquet optimisé

Les exports CSV et JSON sont lents à interroger et peu pratiques à partager. Parquet optimisé est plus rapide à filtrer et à streamer, et les téléversements et téléchargements sont plus rapides grâce à la déduplication Xet. Dans cet exemple, chaque fichier CSV, JSON ou Parquet téléversé dans un bucket apparaît en Parquet optimisé dans un autre. Le Job exécute optimize-parquet.py, un script prêt à l’emploi issu de uv-scripts.

Vous avez besoin d'un compte Hugging Face avec des crédits prépayés et de la CLI hf.

Créer deux buckets

Un bucket reçoit vos téléversements et l'autre reçoit les fichiers Parquet :

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

Créer le Job

Créez le Job que le webhook exécutera :

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

Le script s’exécute depuis son URL, avec hf jobs run. hf jobs uv run est la méthode habituelle pour exécuter un script UV, mais elle télécharge le script sur un volume, ce qu’une exécution via webhook n’aurait pas. cpu-upgrade suffit pour ce travail, et le délai d’expiration de deux heures laisse de la marge pour les fichiers volumineux.

Le script peut également se trouver dans un bucket ou un dépôt privé. N’utilisez pas le bucket surveillé : le téléversement du script déclencherait un Job. Ajoutez --secrets HF_TOKEN à la commande, avec un jeton capable de lire le script. Si l’exécution échoue avec une SyntaxError, le jeton ne peut pas le lire. Le jeton du webhook a besoin du même accès, car les exécutions via webhook ne reçoivent pas ce secret.

hf jobs run démarre le Job immédiatement. Cette première exécution ne génère aucun événement webhook, et optimize-parquet.py ne fait rien sans celui-ci, de sorte que le Job s’arrête aussitôt. Copiez l’identifiant du Job qu’il affiche.

Créer le webhook

Surveillez le premier bucket et exécutez votre Job lorsqu'il change :

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

--domain repo restreint le webhook aux modifications de fichiers et de paramètres. Les buckets n’envoient que ces événements, donc cette option est explicite plutôt que nécessaire, mais elle garantit que la commande reste correcte si vous surveillez plus tard un modèle ou un jeu de données.

--secrets HF_TOKEN stocke un jeton avec le webhook, chiffré. Chaque Job démarré par le webhook le reçoit en tant que secret et l’utilise pour lire et écrire dans les buckets. La valeur provient de HF_TOKEN dans votre environnement, ou du jeton utilisé pour la connexion. Un jeton à granularité fine ne disposant que des permissions nécessaires au Job est le choix le plus sûr.

Pour utiliser un jeton créé spécifiquement pour ce pipeline, passez la valeur explicitement avec --secrets HF_TOKEN=hf_…, ou évitez de l’enregistrer dans l’historique de votre shell en le transmettant via un tube : printf 'HF_TOKEN=hf_…\n' | hf webhooks create … --secrets-file -.

La commande affiche l’identifiant du webhook. Pour arrêter le pipeline plus tard, supprimez le webhook avec hf webhooks delete <webhook ID>.

Téléverser un fichier

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

Environ une minute plus tard, un Job apparaît sur votre page Jobs. S’il n’apparaît pas, ouvrez l’onglet Activity du webhook dans vos paramètres de webhook pour consulter la livraison, et lisez la sortie d’un Job avec hf jobs logs <job ID>. Une fois le Job terminé, le fichier Parquet se trouve dans le second bucket :

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

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

Pour convertir de nombreux fichiers, téléversez-les avec une seule commande hf buckets sync. Les fichiers téléversés ensemble arrivent généralement en tant qu'un seul événement, de sorte qu'un seul Job les convertit tous, jusqu'à 10 000 fichiers par événement. Consultez Webhooks pour la charge utile complète du bucket et ce qui se passe au-delà de cette limite.

Ce que fait le script

La partie spécifique aux webhooks est la lecture de l'événement :

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)

Pour chaque nouveau fichier, le script le charge avec datasets directement depuis le bucket (hf://buckets/...). Si le fichier est plus grand qu'un tiers de l'espace disque libre du Job, il est diffusé en flux au lieu d'être téléchargé. push_to_hub l'écrit ensuite dans le bucket de sortie au format Parquet, avec un fractionnement défini par le contenu, un index de page et des groupes de lignes d'au plus 100 Mo.

Utilisez le modèle pour votre propre tâche

Conservez la configuration et modifiez le script. Par exemple :

• Supprimez les données personnelles avant l’entraînement. Collectez le texte brut dans un bucket privé, masquez les informations personnelles à l’aide d’un modèle tel que GLiNER2 PII filter, puis n’envoyez que le texte masqué dans votre jeu de données d’entraînement.
• Évaluer les nouveaux points de contrôle. Un Job d’entraînement enregistre les points de contrôle dans un bucket, et chaque nouveau point de contrôle déclenche un Job d’évaluation. Si un point de contrôle est téléversé en plusieurs parties, ne réagissez que lorsqu’un fichier marqueur que vous écrivez en dernier apparaît dans updatedFiles ; les événements peuvent arriver dans le désordre.
• Transcrire ou intégrer de nouveaux fichiers. Transformez de nouveaux enregistrements en transcriptions, ou de nouveaux documents en intégrations pour la recherche.

Lorsque vous rédigez votre script, définissez ce qu’il fait lors de la première exécution, lorsque WEBHOOK_PAYLOAD n’est pas défini. Il peut se terminer, comme le fait optimize-parquet.py, ou effectuer un travail réel, tel que le traitement des fichiers déjà présents dans le bucket. Si la première exécution nécessite un jeton, par exemple pour lire le bucket ou un script privé, ajoutez --secrets HF_TOKEN à hf jobs run. Les exécutions par webhook ne reçoivent pas ce secret. Elles obtiennent leur jeton depuis le webhook.

Webhook ou planification ?

Un webhook démarre un Job pour chaque événement, de sorte que les nouveaux fichiers sont traités environ une minute après leur arrivée. S’ils n’ont pas besoin d’être traités aussi rapidement, un scheduled Job constitue une alternative : il s’exécute chaque heure ou chaque jour et traite tout ce qui est arrivé depuis la dernière exécution. Cela regroupe le travail en un nombre réduit de Jobs, ce qui est utile pour les tâches GPU lentes à démarrer, et évite la limite de webhook de 1 000 déclenchements par 24 heures. Le script doit alors trouver lui-même les nouveaux fichiers, par exemple en comparant les buckets d’entrée et de sortie.