如何在unittest中测试气流dag？

Question

我正在尝试在测试环境中测试具有多个任务的 dag。我能够测试与 dag 关联的单个任务，但我想在 dag 中创建多个任务并启动第一个任务。 为了在 dag 中测试一项任务，我正在使用

task1.run()

正在执行。但是，当我在一个 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(2015, 6, 1),
'email': ['airflow@example.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('tutorial', default_args=default_args)

# t1, t2 and t3 are examples of tasks created by instantiating operators
t1 = BashOperator(
task_id='print_date',
bash_command='date',
dag=dag)

t2 = BashOperator(
task_id='sleep',
bash_command='sleep 5',
retries=3,
dag=dag)

t2.set_upstream(t1)

t1.run() # It is executing just first task.

为了运行第二个任务，我必须使用 t2.run() 来运行，因为我在设计 DAG 时不希望这样做。如何做到这一点？

Answer 1

我还不能完全确定我是否理解你的问题，但我会尝试开始回答。

如果您的目标是手动运行 DAG 或其任务的子集，您可以通过 CLI 实现此目的，例如：

• $ airflow run ... - 运行一个任务实例
• $ airflow test ... - 在不检查依赖关系或在数据库中记录状态的情况下测试任务实例
• $ airflow trigger_dag ... - 触发 DAG 的特定 DAG 运行

CLI 文档 - https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html

我认为气流运行命令是与您的用例最相关的命令。

在运行时，在 DAG 中调度任务并在满足其要求后运行下游依赖项均由执行程序自动处理。您不需要在代码中的任何位置调用 run()。

至于run方法本身，代码还在：

• DAG.run(...) - https://github.com/apache/incubator-airflow/blob/985a433a8d03906edfa49a99aa86c026769aff57/airflow/models.py#L4061-L4076
• TaskInstance.run(...) - https://github.com/apache/incubator-airflow/blob/985a433a8d03906edfa49a99aa86c026769aff57/airflow/models.py#L1711-L1722

问题

1. 当您说“在测试环境中”测试 DAG 时，您的确切意思是什么？比如在 CI 或单元测试中？
2. 此代码是用于测试还是来自您的实际 DAG 之一？
3. 这与您最近的另一个问题Test Dag run for Airflow 1.9 in unittest 有关吗？