Airflow "Ce DAG n'est pas disponible dans l'objet DagBag du serveur Web"

Question

lorsque j'ai mis un nouveau DAG script python dans le dags dossier, je peux afficher une nouvelle entrée de DAG dans le groupe de disponibilité de l'INTERFACE utilisateur, mais il n'était pas activé automatiquement. En plus de cela, il semble ne pas chargé correctement. Je ne peux que cliquer sur le bouton Actualiser plusieurs fois sur le côté droit de la liste et faites basculer le bouton on/off sur le côté gauche de la liste pour être en mesure de planifier la DAG. Ce sont le processus manuel que j'ai besoin de déclencher quelque chose, même si le groupe de Script a été mis à l'intérieur de le groupe dossier de.

Quelqu'un peut m'aider sur ce point ? Ai-je raté quelque chose ? Ou est-ce un comportement correct dans la circulation de l'air ?

Par la voie, comme mentionné dans le titre du post, il y a un indicateur avec ce message: "Cette DAG n'est pas disponible dans le serveur web DagBag objet. Il montre dans cette liste parce que le planificateur marqué comme actif dans la metdata" base de données taggés avec le DAG titre avant de me déclencher tout ce processus manuel.

Answer 1

Ce n'est pas vous, ni qu'il est correct ou le comportement attendu. C'est un " bug " avec le flux d'Air. Le serveur web est mise en cache de la DagBag dans une manière que vous ne pouvez pas vraiment l'utiliser comme prévu.

"Attempt removing DagBag caching for the web server" reste sur l'officiel TODO dans le cadre de la feuille de route, indiquant que ce bug n'est pas encore complètement résolu, mais voici quelques suggestions sur la façon de procéder:

utilisez uniquement des constructeurs d'air v1.9+

Avant air v1.9 cela se produit lorsqu'un groupe est instancié par une fonction qui est importé dans le fichier où l'instanciation se passe. Qui est: quand un constructeur ou de l'usine modèle est utilisé. Certains rapports de cette question sur github 2 et JIRA 3 conduit à un correctif publié dans air v1.9.

Si vous utilisez une ancienne version de la circulation de l'air, ne pas utiliser de générateur de fonctions.

airflow backfill pour recharger le cache

Comme Dmitri suggère, l'exécution d' airflow backfill '<dag_id>' -s '<date>' -e '<date>' pour le même point de départ et une date de fin peut parfois aider. Par la suite, vous pouvez vous retrouver avec de la (non)-problème de Priyank points, mais c'est le comportement attendu (état: pause ou pas) en fonction de la configuration que vous avez dans votre installation.

Answer 2

Redémarrez le airflow webserver résout mon problème.

Answer 3

Le problème vient du fait que le DAG par défaut est mis en pause dans le DagBag afin que le planificateur ne soit pas submergé par de nombreuses activités de renvoi au démarrage / au redémarrage.

Pour contourner ce problème, modifiez le paramètre ci-dessous dans votre fichier airflow.cfg :

 # Are DAGs paused by default at creation 
dags_are_paused_at_creation = False
 

J'espère que cela t'aides. À votre santé!

Answer 4

J'ai une théorie sur la cause possible de ce problème dans Google Compositeur. Il y a la section à propos de dag échecs sur le serveur web de documentation de dépannage pour le Compositeur, qui dit:

Éviter de courir des poids lourds de calcul à DAG moment de l'analyse. Contrairement à l' travailleur et planificateur de nœuds, dont les types de machines peuvent être personnalisés pour avoir une plus grande CPU et de la capacité de la mémoire, le serveur utilise un fixe type de machine, ce qui peut conduire à DAG analyse des échecs si l'analyse en temps le calcul est trop lourd.

Et j'ai essayé de charger la configuration à partir de la source externe (qui s'est réellement négligeable de temps à le comparer à d'autres opérations pour créer DAG, mais encore cassé quelque chose, parce que le serveur web de la circulation de l'Air dans compositeur s'exécute sur App Engine, qui a des comportements étranges).

J'ai trouvé la solution de contournement dans la discussion de ce problème lié à Google, et il est de créer des DAG de la tâche qui charge toutes les données nécessaires et les stocke les données dans le flux d'air variable:

Variable.set("pipeline_config", config, serialize_json=True)

Ensuite, j'ai pu faire

Variable.get("pipeline_config", deserialize_json=True)

Et réussir à générer pipeline à partir de cela. Avantage supplémentaire est que je reçois des journaux de cette tâche, qui-je obtenir à partir du serveur web, à cause de ce problème.