Le job Cloud Run de traitement en lot

Le script nocturne d'origine traitait les avis un par un, dans une seule boucle : 31 appels Gemini à la suite, 4 minutes, et une exception au 17e avis qui faisait perdre les 16 premiers puisque rien n'était écrit avant la fin. Pour 30 000 avis, la même boucle prendrait des heures. Il faut découper le travail en tranches indépendantes, traiter chaque tranche en parallèle, regrouper les appels au modèle, et écrire chaque tranche dès qu'elle est finie.

Le mécanisme

Une exécution du job Cloud Run reçoit, par le workflow, le fichier à traiter (INPUT_URI), le préfixe de sortie (OUTPUT_PREFIX) et le nombre de tâches (taskCount). Cloud Run démarre les tâches en parallèle ; chacune ne connaît que son index.

flowchart TB
  F["entrant/2026-10-15/email.jsonl<br/>10 lignes"] --> T0["Tâche 0<br/>lignes 1, 3, 5, 7, 9"]
  F --> T1["Tâche 1<br/>lignes 2, 4, 6, 8, 10"]
  T0 -- "stats.json existe ?" --> S0{"oui : sortie immédiate<br/>non : traiter"}
  S0 --> L0["lots de 5 avis → Gemini<br/>un classement par avis exigé"]
  L0 --> W0["part-000-of-002.jsonl<br/>puis part-000-of-002.stats.json"]
  T1 --> W1["part-001-of-002.jsonl<br/>puis part-001-of-002.stats.json"]

Trois décisions structurent la tâche :

1. Découpage par position (lignes[index::count]) : chaque ligne appartient à exactement une tâche, sans coordination entre tâches. Le fichier d'entrée ne change pas pendant la nuit, la position est donc stable.
2. Lots de 5 avis par appel Gemini, en sortie structurée : un tableau de classements. La consigne et le schéma de sortie sont envoyés une fois par lot au lieu d'une fois par avis. En contrepartie, le modèle peut oublier un avis : la tâche exige autant de classements que d'avis, avec les mêmes id, redemande une fois, puis échoue.
3. Le marqueur écrit en dernier. part-NNN.stats.json n'est écrit qu'après part-NNN.jsonl. Sa présence signifie « tranche terminée » ; une tâche qui le trouve ne fait rien.

Les erreurs se rangent ainsi :

La mise en œuvre

def run_task(store: Store, classify: Classifier, input_uri: str, output_prefix: str,
             index: int, count: int, batch_size: int = 5) -> dict:
    part = f"{output_prefix}/part-{index:03d}-of-{count:03d}"
    done = store.read(f"{part}.stats.json")
    if done is not None:
        log.info("tranche déjà traitée", extra={"fields": {"event": "tranche_ignoree", "part": part}})
        return json.loads(done)

    raw = store.read(input_uri)
    if raw is None:
        raise FileNotFoundError(input_uri)
    lines = [l for l in raw.splitlines() if l.strip()][index::count]
    ...
    for i in range(0, len(avis), batch_size):
        out, used = classify_checked(classify, avis[i:i + batch_size])
        ...
    store.write(f"{part}.jsonl", "".join(c.model_dump_json() + "\n" for c in results), "application/x-ndjson")
    store.write(f"{part}.stats.json", json.dumps(stats, ensure_ascii=False), "application/json")  # en dernier : marqueur
    return stats

Le contrat de sortie, avec la définition de la gravité dans le schéma, là où le modèle la lit :

class Classement(BaseModel):
    id: str = Field(description="L'id de l'avis, recopié à l'identique.")
    theme: Theme
    sentiment: Literal["negatif", "neutre", "positif"]
    gravite: Literal["basse", "moyenne", "haute"] = Field(
        description=("haute = menace de litige ou de plainte, argent du client bloqué (double débit, remboursement promis "
                     "non versé), personne vulnérable laissée sans aide ; moyenne = insatisfaction réelle sans ces enjeux ; "
                     "basse = question, demande simple ou avis positif."))
    action: str = Field(description="Une phrase : ce que le service qualité doit faire.")

Le fichier stats.json est ce que le workflow lit ensuite : des compteurs et la liste des cas graves, quelques centaines d'octets, jamais le texte des avis.

Le job est créé une fois, avec des valeurs par défaut que le workflow remplace à chaque exécution :

JOB_OPTS=(--region="$REGION" --image="$IMAGE" --service-account="$JOB_SA" --tasks=1 --max-retries=2
  --task-timeout=10m --cpu=1 --memory=512Mi --labels="$LABEL"
  --set-env-vars="GOOGLE_CLOUD_PROJECT=$PROJECT,GOOGLE_CLOUD_LOCATION=$REGION,MODEL=${MODEL:-gemini-2.5-flash},BATCH_SIZE=5")

Résultat obtenu (2026-10-16)

Les tests de la tâche, avec un faux Gemini et un stockage en mémoire :

$ make test
..............                                                           [100%]
14 passed in 0.46s

Sur Google Cloud, une exécution de job lancée par le workflow, vue par gcloud run jobs executions list :

   JOB                     EXECUTION                     REGION        RUNNING  COMPLETE  CREATED
✔  lab-batch-nocturne-job  lab-batch-nocturne-job-zw9nw  europe-west1  0        2 / 2     2026-10-16 13:31:07 UTC
✔  lab-batch-nocturne-job  lab-batch-nocturne-job-cq48f  europe-west1  0        2 / 2     2026-10-16 13:31:07 UTC
✔  lab-batch-nocturne-job  lab-batch-nocturne-job-mrp2z  europe-west1  0        2 / 2     2026-10-16 13:31:09 UTC

Un premier essai a aussi mesuré la qualité de la gravité. Avec la définition initiale (« haute = risque juridique, demande de remboursement chiffrée, client en détresse ou menace de litige »), Gemini classait 15 avis sur 29 en gravité haute, dont « le guide parlait seulement anglais » et « frais de dossier de 45 € non annoncés ». Une catégorie qui contient la moitié des avis ne trie plus rien. La définition ci-dessus, qui nomme aussi moyenne et basse, est le résultat de cette mesure ; son effet est montré en leçon 5.

Ce qui casse en production

• Lots trop gros. À 50 avis par appel, le modèle saute plus souvent un avis ou mélange deux id, et un lot redemandé coûte 50 classements. Parade : la vérification des id (elle transforme l'oubli en erreur visible) et une taille de lot choisie sur un jeu de référence.
• Tâche qui dépasse son délai. --task-timeout=10m tue une tâche bloquée sur des 429 en série ; elle est relancée, et refait sa tranche entière. Si les relances échouent toutes, l'exécution échoue et le workflow réessaie (leçon suivante). Parade : surveiller la durée des tâches dans les journaux (duree_ms de l'événement tache_terminee).
• Parallélisme qui sature le quota. Chaque tâche fait ses appels en série, mais N tâches × M jobs simultanés font N × M appels concurrents sur un quota partagé par le projet. Parade : concurrency_limit du workflow, et --parallelism du job si le nombre de tâches augmente.
• Changement du nombre de tâches entre deux relances. Avec TASKS_PER_FILE à 2 puis 3, les anciens part-000-of-002 et les nouveaux part-000-of-003 se mélangeraient dans les statistiques. Parade : le nombre de tâches fait partie du chemin de sortie (…/t2/), chaque découpage a son propre jeu de marqueurs.
• Texte des avis dans les journaux. Un log.info(avis) de débogage copie des données personnelles dans Cloud Logging pour 30 jours. Parade : journaliser des identifiants et des compteurs, jamais le contenu.

La règle de décision

Découpez par une règle que chaque tâche applique seule (position, hachage d'identifiant), regroupez les appels au modèle par lots dont vous vérifiez la complétude, écrivez la sortie d'une tranche avant son marqueur. Une erreur de donnée est écartée et comptée ; une erreur passagère fait échouer la tâche pour qu'elle soit relancée ; une erreur de configuration fait échouer l'exécution.

Pratique

Exercice. Remplacez le découpage par position par un découpage par hachage de l'id de l'avis. Écrivez la ligne, et dites dans quel cas ce choix devient nécessaire.

import zlib
mine = [a for a in avis_valides if zlib.crc32(a.id.encode()) % count == index]

• Je découpe un traitement en tâches indépendantes avec CLOUD_RUN_TASK_INDEX et CLOUD_RUN_TASK_COUNT.
• Je regroupe les appels Gemini en lots et je vérifie qu'aucun élément n'est oublié.
• J'écris la sortie d'une tranche avant son marqueur de fin.
• Je range chaque erreur : donnée écartée, tâche relancée, exécution en échec.