Gestion des OOM-Kills Airflow - limites de mémoire process et fuites cumulatives

Question

Gestion des OOM-Kills Airflow - limites de mémoire process et fuites cumulatives

Gabriel Martin 0

Si l'on baisse simplement worker_concurrency, un worker Celery peut toujours subir des fuites de mémoire cumulées au fil de plusieurs exécutions légères et finir par se faire OOM-kill sur une tâche lourde. Pourquoi ne pas avoir plutôt recommandé d'associer un worker_concurrency réduit à la directive worker_max_tasks_per_child (ou d'isoler cette tâche sur une file Celery dédiée via un pool d'exécution custom), et comment gérez-vous la fragmentation mémoire liée au garbage collector de Python (gc.collect()) sur les processus forkés ?

0 commentaires

1 réponse

Votre réponse

Answer 1

Bonjour Gabriel,

Tu as raison de pointer que réduire simplement worker_concurrency ne suffit pas à contenir les fuites cumulatives. La mémoire consommée par un worker Celery sous Airflow n’est pas uniquement liée au nombre de tâches simultanées, mais aussi au cycle de vie du process. Si un worker reste longtemps en vie, il accumule des fragments mémoire et objets non libérés, ce qui finit par provoquer un OOM-kill dès qu’une tâche lourde arrive.

La directive worker_max_tasks_per_child est précisément conçue pour limiter ce phénomène. En forçant le recyclage du process après un certain nombre de tâches, on évite que les fuites s’accumulent indéfiniment. C’est une pratique recommandée dans les environnements où les DAGs déclenchent des tâches Python longues ou avec des librairies qui ne relâchent pas bien la mémoire (pandas, numpy, etc.). Associer un worker_concurrency réduit à un worker_max_tasks_per_child raisonnable (par exemple 50 ou 100 selon la charge) est bien plus efficace que de jouer uniquement sur la concurrence.

Concernant l’isolation des tâches lourdes, créer une queue Celery dédiée avec un pool d’exécution séparé est une approche robuste. Cela permet de réserver des workers avec des limites mémoire plus strictes ou un max_tasks_per_child plus bas, sans impacter les tâches légères. Dans Airflow, tu peux déclarer une queue spécifique dans airflow.cfg et router les DAGs concernés via queue="heavy_tasks" dans l’opérateur.

Sur la fragmentation mémoire et le garbage collector Python, il faut être lucide : appeler gc.collect() manuellement sur des workers forkés n’a qu’un effet limité. Le GC ne résout pas la fragmentation native du heap C, et certaines bibliothèques externes ne libèrent pas correctement leurs buffers. Le recyclage de process reste la seule méthode fiable. Si tu observes des fuites persistantes, tu peux aussi activer worker_prefetch_multiplier=1 pour limiter la rétention d’objets en mémoire entre les tâches, mais cela a un impact sur le throughput.

En résumé, la combinaison optimale est : réduire worker_concurrency pour limiter la pression simultanée, définir worker_max_tasks_per_child pour casser les fuites cumulatives, et isoler les workloads lourds sur une queue dédiée. Le GC manuel n’est pas une solution durable, le recyclage de process est la meilleure pratique.HP.