Webhook kılavuzu: Jobs ile bir kovadaki yeni dosyaları işleme

Storage Buckets, zamanla gelen dosyalar için iyi bir yerdir: diğer araçlardan dışa aktarımlar, günlükler, kayıtlar, taramalar. Bu dosyaların kullanışlı hale gelmeden önce üzerinde yapılması gereken bir iş akışınız genellikle olur; örneğin dönüştürme, temizleme veya transkripsiyon.

Kova üzerinde bir webhook ve bir Job, bir sunucu olmadan bu işi otomatik olarak yapabilir. Dosyalar değiştiğinde webhook, Job'u başlatır. Job, değişen dosyaların listesini alır, onları işler ve durur. Yalnızca çalıştığı süre boyunca ödeme yaparsınız.

Bu kılavuz, deseni açıklar ve ardından tek bir örnekle kurulumunu gösterir: yüklenen CSV ve JSON dosyalarını optimize edilmiş Parquet'e dönüştürme.

Nasıl çalışır

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

Bir webhook, bir URL'yi çağırmak yerine bir Job'u tetikleyebilir. Job'u bir kez oluşturursunuz ve her webhook olayı, olayı ortamında barındırarak onu tekrar çalıştırır. Bir bucket için WEBHOOK_PAYLOAD, değişen dosyaları listeler:

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

Bu sayede İş yalnızca yeni dosyaları işler ve herhangi bir şey indirmeden önce her birini nasıl ele alacağına karar verebilir.

Her çalıştırma, İş'in (Job) görüntüsünü, komutunu, ortam değişkenlerini, donanım türünü ve zaman aşımını yeniden kullanır. İş'in (Job) birimlerini (volumes) veya gizli bilgilerini (secrets) almaz. Bu nedenle aşağıdaki adımlar, betiği bir URL'den çalıştırır ve belirteci (token) webhook'a yerleştirir.

Job, sonuçlarını ikinci kovaya yazar. İzlediği kovaya yazsaydı, kendi çıktısı webhook'u tekrar tetiklerdi.

Örnek: Yüklemeleri optimize edilmiş Parquet'e dönüştürme

CSV ve JSON dışa aktarımları sorgulamak yavaştır ve paylaşımı zahmetlidir. Optimized Parquet, filtreleme ve akış işlemleri için daha hızlıdır ve Xet yinelenme giderme sayesinde daha hızlı yüklenir ve indirilir. Bu örnekte, bir kovaya yüklenen her CSV, JSON veya Parquet dosyası başka bir kovada optimize Parquet olarak görünür. İş, uv-scripts kaynaklı hazır bir bet olan optimize-parquet.py betiğini çalıştırır.

Ön ödemeli kredileri olan bir Hugging Face hesabına ve hf CLI'na ihtiyacınız var.

İki varlık oluşturun

Bir bucket yüklemelerinizi, diğeri Parquet dosyalarını alır:

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

İşi Oluştur

Webhook'un çalıştıracağı Job'ı oluşturun:

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

Betik, hf jobs run ile URL'sinden çalıştırılır. hf jobs uv run, bir UV betiğini çalıştırmak için yaygın yöntemdir ancak betiği bir hacme yükler; webhook çalıştırması bunun için uygun değildir. Bu iş için cpu-upgrade yeterlidir ve iki saatlik zaman aşımı, büyük dosyalar için alan bırakır.

Betik, özel bir bucket veya repo içinde de olabilir. İzlenen bucket’ı kullanmayın: betiğin yüklenmesi bir Job başlatır. Betiği okuyabilen bir token ile komuta --secrets HF_TOKEN ekleyin. Çalıştırma SyntaxError ile başarısız olursa, token betiği okuyamıyordur. Webhook’un token’ı da aynı erişime ihtiyaç duyar, çünkü webhook çalıştırmaları bu gizli bilgiyi almaz.

hf jobs run komutu Job'u hemen başlatır. Bu ilk çalıştırmada webhook olayı yoktur ve optimize-parquet.py bir olay olmadan hiçbir şey yapmaz, bu nedenle Job hemen durur. Yazdırdığı Job ID'sini kopyalayın.

Webhook'ü oluşturun

İlk kovayı izleyin ve değiştiğinde Job'unuzu çalıştırın:

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

--domain repo webhook'ü dosya ve ayar değişiklikleriyle sınırlar. Buckets yalnızca bu tür değişiklikleri gönderir, bu yüzden bu ayar zorunlu olmaktan çok açıklayıcıdır; ancak ileride bir model veya veri setini izlemeye başlarsanız komutun doğru kalmasını sağlar.

--secrets HF_TOKEN, bir webhook ile şifrelenmiş olarak bir token saklar. Webhook’un başlattığı her Job, bunu bir secret olarak alır ve bucket’ları okumak ve yazmak için kullanır. Değer, ortamınızdaki HF_TOKEN’dan veya oturum açarken kullandığınız token’dan gelir. Job’ın ihtiyaç duyduğu izinlere sahip bir fine-grained token en güvenli seçenektir.

Bu pipeline için özel olarak oluşturulmuş bir token kullanmak için, değeri --secrets HF_TOKEN=hf_… ile açıkça iletin veya kabuk geçmişinizde tutmamak için boru hattıyla iletin: printf 'HF_TOKEN=hf_…\n' | hf webhooks create … --secrets-file -.

Komut, webhook kimliğini yazdırır. Boru hattını daha sonra durdurmak için webhook'u hf webhooks delete <webhook ID> ile silin.

Bir dosya yükle

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

Yaklaşık bir dakika sonra, Jobs sayfanızda bir Job görünür. Görünmüyorsa, teslimatı görmek için webhook ayarlarınızda webhook'un Activity sekmesini açın ve bir Job'un çıktısını hf jobs logs <job ID> ile okuyun. Job bittiğinde, Parquet dosyası ikinci kovada bulunur:

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

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

Çok sayıda dosyayı dönüştürmek için onları tek bir hf buckets sync ile yükleyin. Birlikte yüklenen dosyalar genellikle tek bir olay olarak gelir, bu nedenle tek bir Job, olay başına 10.000 dosyaya kadar hepsini dönüştürür. Tam bucket yükü ve bu sınırın üzerinde ne olduğunu görmek için Webhooks bölümüne bakın.

Betik ne yapar

Webhook'lara özgü olan kısım, olayı okumaktır:

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)

Her yeni dosya için betik, dosyayı doğrudan kovadan datasets ile yükler (hf://buckets/...). Dosya, İşin boş disk alanının üçte birinden büyükse indirilmek yerine akış olarak işlenir. Ardından push_to_hub, dosyayı içerik tanımlı parçalama, sayfa indeksi ve en fazla 100 MB'lık satır grupları içeren Parquet biçiminde çıktı kovasına yazar.

Kendi göreviniz için deseni kullanın

Kurulumu koruyun ve betiği değiştirin. Örneğin:

• Eğitimden önce kişisel verileri kaldırın. Ham metni özel bir kovaya toplayın, GLiNER2 PII filter gibi bir modelle kişisel bilgileri sansürleyin ve yalnızca sansürlenmiş metni eğitim veri setinize gönderin.
• Yeni checkpoint'leri değerlendirin. Bir eğitim Job'ı checkpoint'leri bir bucket'a kaydeder ve her yeni checkpoint bir değerlendirme Job'ı başlatır. Bir checkpoint birden fazla parça halinde yükleniyorsa, yalnızca son olarak yazdığınız işaretçi dosyası updatedFiles içinde göründüğünde tepki verin; olaylar sırasız olarak gelebilir.
• Yeni dosyaları transkribe edin veya gömme (embedding) oluşturun. Yeni kayıtları transkriptlere, yeni belgeleri ise arama için gömme (embedding) vektörlerine dönüştürün.

Betiklerinizi yazarken, WEBHOOK_PAYLOAD ayarlı değilken ilk çalıştırmada ne yapacaklarına karar verin. optimize-parquet.py'nin yaptığı gibi çıkabilirler veya bucket'ta zaten bulunan dosyaları işleme gibi gerçek işler yapabilirler. İlk çalıştırma, örneğin bucket'ı veya özel bir betiği okumak için bir jeton gerektiriyorsa, hf jobs run komutuna --secrets HF_TOKEN ekleyin. Webhook çalıştırmaları bu sırrı almaz. Jetonlarını webhook'tan alırlar.

Webhook mu yoksa zamanlama mı?

Webhook, her olay için bir Job başlatır; bu nedenle yeni dosyalar, varışlarından yaklaşık bir dakika sonra işlenir. Bu kadar hızlı işlenmeleri gerekmiyorsa, zamanlanmış Job bir alternatiftir: saatte bir veya günde bir çalışır ve son çalışmadan bu yana gelen her şeyi işler. Bu, işleri daha az sayıda Job’a gruplandırır; bu da başlatılması yavaş olan GPU görevlerine yardımcı olur ve 24 saatte 1.000 tetikleme sınırına takılmayı önler. Bu durumda, örneğin giriş ve çıkış varillerini karşılaştırarak, yeni dosyaları bulma işi script’e düşer.