如何运行简单的 Airflow DAG答案

【问题标题】：How to Run a Simple Airflow DAG如何运行简单的 Airflow DAG
【发布时间】：2017-06-07 21:34:19
【问题描述】：

我对 Airflow 完全陌生。我想在指定日期运行一个简单的 DAG。我正在努力区分开始日期、执行日期和回填。运行 DAG 的命令是什么？

这是我从那以后尝试过的：

airflow run dag_1 task_1 2017-1-23

我第一次运行该命令时，任务正确执行，但当我再次尝试时，它没有工作。

这是我运行的另一个命令：

airflow backfill dag_1 -s 2017-1-23 -e 2017-1-24

我不知道这个命令会发生什么。 DAG 会在每天 23 点到 24 点执行吗？

在运行上面的两个命令之前，我这样做了：

airflow initdb
airflow scheduler 
airflow webserver -p 8085 --debug &

这是我的 DAG

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2017, 1, 23, 12),
    'email': ['airflow@airflow.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'dag_1', default_args=default_args, schedule_interval=timedelta(1))

t1 = BashOperator(
    task_id='create_clients',
    bash_command='Rscript /scripts/Cli.r',
    dag=dag)

t2 = BashOperator(
    task_id='create_operation',
    bash_command='Rscript Operation.r',
    retries=3,
    dag=dag)

t2.set_upstream(t1)

截图：Tree View

更新

airflow run dag_1 task_1 2017-1-23T10:34

【问题讨论】：

如果你真的尝试过，请edit你的问题包括minimal reproducible example。

标签： python airflow

【解决方案1】：

如果你用

运行一次

airflow run dag_1 task_1 2017-1-23

运行被保存并且再次运行它不会做任何事情你可以尝试通过强制它重新运行它

airflow run --force=true dag_1 task_1 2017-1-23

气流回填命令将运行在从开始日期到结束日期指定的时间段内运行的任何执行。这取决于您在 DAG 上设置的计划，如果您将其设置为每小时触发一次，它应该运行 24 次，但它也不会重新执行之前执行的运行。

您可以清除任务，就像它从未运行过一样

airflow clear dag_1 -s 2017-1-23 -e 2017-1-24

还可以在此处查看 cli 文档：https://airflow.incubator.apache.org/cli.html

【讨论】：

感谢您的解释。我试图增加第一次运行的时间（见更新）。为什么即使没有达到指定的时间，运行也会立即执行？例如，我的当前时间是 10:30，我在运行中指定 10:34...它立即运行，这是正常行为吗？
我相信（不是 100% 肯定）它运行任务就像它是指定的日期，因为你将它作为参数传入。因此，当它完成时，保存的有关运行的信息表明它当时完成了该任务的运行。
-f 选项即可，无需设置=true。气流 1.9 airflow run: error: argument -f/--force: ignored explicit argument 'true'

【解决方案2】：

开始日期、执行日期和回填之间的差异

完成回填以显式运行 DAG 以测试/手动运行 DAG/重新运行出错的 DAG。您可以使用 CLI 完成此操作

airflow backfill -s <<start_date>> <<dag>> 
#optionally provide -1 as start_date to run it immediately

start_date，顾名思义，是 DAG 定义有效的日期

execution_date 是要运行的日期时间。这是您在测试 DAG 的单个任务时提供的，如下所示

airflow test <<dag>> <<task>> <<exec_date>>

运行 dag 的命令是什么

Backfill 是显式运行 DAG 的命令。否则，您只需将 DAG 放在 DAGBAG 文件夹中，调度程序就会按照 DAG 定义中定义的调度运行它

airflow backfill -s <<start_date>> <<dag>> 
#optionally provide -1 as start_date to run it immediately

【讨论】：