无法在 Airflow 中部署 DAG

Question

我无法在气流中部署 DAG。不断给出相同的错误。运行在 8080 端口上，调度程序也已打开。

Running %s on host %s <TaskInstance: bworkflow_dag.bworkflow_template 2020-08-11T00:00:00+00:00 [failed]> airflow-instance-test.c.cc-data-sandbox.internal
[2020-08-11 14:46:29,030] {__init__.py:50} INFO - Using executor SequentialExecutor
[2020-08-11 14:46:29,031] {dagbag.py:396} INFO - Filling up the DagBag from /home/kshitij/airflow/dags
/home/kshitij/.local/lib/python3.5/site-packages/airflow/models/dag.py:1342: PendingDeprecationWarning: The requested task could not be added to the DAG because a task with task_id create_tag_template_field_result is already in the DAG. Starting in Airflow 2.0, trying to overwrite a task will raise an exception.
  category=PendingDeprecationWarning)
Running %s on host %s <TaskInstance: bworkflow_dag.bworkflow_template 2020-08-11T00:00:00+00:00 [failed]> airflow-instance-test.c.cc-data-sandbox.internal

我使用的路径正确，即 ~/airflow/dags

找到代码片段：

from builtins import range
from datetime import timedelta
from airflow.models import DAG
from airflow.utils.dates import days_ago
from airflow.contrib.operators.dataproc_operator import DataprocWorkflowTemplateInstantiateOperator

args = {
    'owner': 'Airflow',
    'start_date': days_ago(2),
}

dag = DAG(
    dag_id='workflow_dag',
    default_args=args,
    schedule_interval=None,
    dagrun_timeout=timedelta(days=1),
)

workflow_template = DataprocWorkflowTemplateInstantiateOperator(
    template_id="workflow_rds",
    project_id="<project name>",
    task_id="workflow_template",
    dag=dag)


workflow_template


if __name__ == "__main__":
    dag.cli()

这是一个单任务 DAG。

让我知道哪里出错了。

Answer 1

正如在评论中讨论的那样，以下几行不是必需的，需要删除

...
workflow_template


if __name__ == "__main__":
    dag.cli()

接受他们one-by-one

workflow_template：这什么都不做；只是引用一个包含 task 的变量（删除它只是清理）
if __name__ == "__main__":：仅当文件为executed as a main program时才执行某段代码。 DAG 文件中不需要
dag.cli(): 我相信这是 cuprit。我以前没有使用过这种方法，但是 docstring 说它 Exposes a CLI specific to this DAG，这给解析 dag-definition 文件的过程带来了问题（实际上是 webserver）

无法在 Airflow 中部署 DAG

Unable to deploy a DAG in Airflow

google-cloud-dataproc

airflow