使用 Airflow UI 执行整个 DAG答案

【问题标题】：Execute entire DAG using Airflow UI使用 Airflow UI 执行整个 DAG
【发布时间】：2017-04-13 00:58:48
【问题描述】：

我是气流的新手，我们有一个有 3 个任务的 DAG。目前我们正在使用 Celery Executor，因为我们需要灵活地运行单个任务。我们不想安排工作流程，现在它将是手动触发。有什么方法可以使用 Airflow UI 来执行整个工作流程（和我们在 oozie 中的一样）？

一次执行一项任务很痛苦。

【问题讨论】：

澄清一下，您是否希望在运行第一个任务时运行所有三个任务？也请发布您的相关代码。
如果您设置了依赖关系，然后使用airflow trigger_dag id 从命令行运行 dag，有什么问题？
是的，你没看错。我们需要在运行第一个任务时运行所有任务。抱歉，由于限制，无法发布代码。我们将把 DAG 交给支持团队，他们的工作是手动触发工作流。由于他们对命令行没有太多经验，我们需要通过 UI 执行它。
在单独的线程中运行 airflow schedule（在 Web 服务器旁边）。然后你就可以手动触发 DAG

标签： python airflow

【解决方案1】：

在 Airflow 1.8 及更高版本中，仪表板上的每个 dag 都有一个按钮，看起来像一个播放按钮：

在旧版本的 Airflow 中，您可以使用位于以下位置的对话框：

Browse -> Dag Runs -> Create

任何一个都应该从 UI 开始。

【讨论】：

这是否允许您将参数传递给 dag？
这不是让你只添加一个运行，而不是一个范围吗？

【解决方案2】：

我会尝试一下，希望你能调整你的代码来处理这个问题。

default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2015, 6, 1),
'email': ['airflow@airflow.com'],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
# 'queue': 'bash_queue',
# 'pool': 'backfill',
# 'priority_weight': 10,
# 'end_date': datetime(2016, 1, 1),
}

dag = DAG('your_dag', default_args=default_args)

#start of your tasks

first_task = BashOperator(task_id='print_date',
                          bash_command='python script1_name args',
                          dag=dag)
second_task = BashOperator(task_id='print_date',
                          bash_command='python script2_name args',
                          dag=dag)
third_task = BashOperator(task_id='print_date',
                          bash_command='python script_name args',
                          dag=dag)

#then set the dependencies
second_task.set_upstream(first_task)
third_task.set_upstream(second_task)

然后当你触发 DAG 时，所有三个任务都会按顺序运行。如果这些任务不相互依赖，您可以从脚本中删除set_upstream() 行。请注意，所有这些任务必须在同一个脚本中才能使用一个命令运行。

【讨论】：

感谢您为解答所花费的时间和精力。但同样如前所述，我想通过 Airflow 仪表板运行它。我不想通过命令提示符或使用调度机制来触发此工作流。是否有任何机制，例如单击它的按钮，它将按提到的顺序运行 dag。？
另外，你可能还需要导入一些东西：``` from airflow import DAG from airflow.operators.bash_operator import BashOperator from datetime import datetime, timedelta ```