文件建议对DAG文件进行基本的健全性检查是对其进行解释。即:
$ python ~/path/to/my/dag.py
我发现这很有用。但是,现在我已经在$aifflow\u HOME/plugins
下创建了一个插件,MordorOperator
:
from airflow.plugins_manager import AirflowPlugin from airflow.utils.decorators import apply_defaults from airflow.operators import BaseOperator from airflow.exceptions import AirflowException import pika import json class MordorOperator(BaseOperator): JOB_QUEUE_MAPPING = {"testing": "testing"} @apply_defaults def __init__(self, job, *args, **kwargs): super().__init__(*args, **kwargs) # stuff def execute(self, context): # stuff class MordorPlugin(AirflowPlugin): name = "MordorPlugin" operators = [MordorOperator]
我可以导入插件,并在一个示例DAG中看到它的工作:
from airflow import DAG from airflow.operators import MordorOperator from datetime import datetime dag = DAG('mordor_dag', description='DAG with a single task', start_date=datetime.today(), catchup=False) hello_operator = MordorOperator(job="testing", task_id='run_single_task', dag=dag)
然而,当我试图解释这个文件时,我得到了失败,我怀疑我不应该得到,因为插件成功运行。我的怀疑是,这是因为在运行时发生了一些动态代码生成,而当DAG被自己解释时,这些代码生成是不可用的。我还发现PyCharm在导入插件时不能执行任何自动完成。
(venv) 3:54PM /Users/paymahn/solvvy/scheduler mordor.operator ✱ ❮❮❮ python dags/mordor_test.py section/key [core/airflow-home] not found in config Traceback (most recent call last): File "dags/mordor_test.py", line 2, in from airflow.operators import MordorOperator ImportError: cannot import name 'MordorOperator'
如何对使用插件的DAG进行健全性测试?是否可以让PyCharm为自定义运算符提供自动完成功能?
我在docker容器中运行airflow,有一个脚本作为容器入口点运行。当我运行测试时,plugins
文件夹对我的容器不可用。作为安装脚本的一部分,我必须在容器中添加一个符号链接。我的问题的解决方案对我来说是非常具体的,如果其他人偶然发现了这一点,我对你的情况没有一个好的答案,除了:确保你的插件文件夹正确可用。