python - 在一个 DAG 中执行顺序和并发任务

Question

我是 Airflow 新手，有一些关于如何在一个 DAG 中同时正确运行某些任务和顺序运行其他任务的一些基本问题。

在我的 DAG 中，基本步骤是:刷新数据、运行 3 个单独的脚本、部署。这些应用程序中的每一个都在单独的 Docker 容器中运行。

在下面的示例中，一切都是按顺序完成的，但是，我的目标是刷新数据，然后并行执行这个、那个和其他事物，然后部署。

refresh >> [this, that, the_other_thing] >> deploy

我只想在 [this, that, the_other_thing] 之后部署已经完成，但尚不清楚三者中哪一个会最后完成。在一个 DAG 中执行此序列的最佳实践是什么？设置concurrency=3就够了吗并执行[this, that, the_other_thing]在 for loop ？任何建议都表示赞赏

from builtins import range
from datetime import timedelta, datetime

from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago
from airflow.hooks.base_hook import BaseHook

image = 'myserver.com:8080/my_project:latest'

args = {
    'owner': 'Airflow',
    'start_date': datetime(2020,01,01),
    'depends_on_past': False,
    "retries": 2,
    'retry_delay': timedelta(minutes=5)
}

conn_foo_db = BaseHook.get_connection('foobar_db')
conn_docker = BaseHook.get_connection('my_registry')

dag = DAG(
    dag_id='analysis',
    default_args=args,
    schedule_interval='0 3 * * *',
    dagrun_timeout=timedelta(minutes=180),
    max_active_runs=1,
    concurrency=1,
    tags=['daily']
)

refresh_data = BashOperator(
    task_id='refresh_data',
    bash_command='docker run '
                 '-i --rm '
                 f"-e DB_PASSWORD='{ conn_foo_db.password }' "
                 f' { image }  '
                 'app=refresh',
    dag=dag,
)

this = BashOperator(
    task_id='run_app_this',
    bash_command='docker run '
                 '-i --rm '
                 f"-e DB_PASSWORD='{ conn_foo_db.password }' "
                 f' { image }  '
                 'app=do_this ',
    dag=dag,
)

that = BashOperator(
    task_id='run_app_that',
    bash_command='docker run '
                 '-i --rm '
                 f"-e DB_PASSWORD='{ conn_foo_db.password }' "
                 f' { image }  '
                 'app=do_that',
    dag=dag,
)

the_other_thing = BashOperator(
    task_id='run_app_the_other_thing',
    bash_command='docker run '
                 '-i --rm '
                 f"-e DB_PASSWORD='{ conn_foo_db.password }' "
                 f' { image }  '
                 'app=do_the_other_thing ',
    dag=dag,
)

deploy = BashOperator(
    task_id='deploy',
    bash_command='docker run '
                 '-i --rm '
                 f"-e DB_PASSWORD='{ conn_foo_db.password }' "
                 f' { image }  '
                 'app=deploy ',
    dag=dag,
)

refresh_data >> run_app_this >> run_app_that >> run_app_the_other_thing >> deploy_to_dashboard

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

Answer 1

是的，你的假设是正确的。 可能的代码可以是:

tasks_list = ["this", "that", "the_other_thing"]

refresh_data = BashOperator(
    task_id='refresh_data_task',
    bash_command='docker run '
                 '-i --rm '
                 f"-e DB_PASSWORD='{ conn_foo_db.password }' "
                 f' { image }  '
                 'app=refresh',
    dag=dag,
)

deploy = BashOperator(
    task_id='deploy_task',
    bash_command='docker run '
                 '-i --rm '
                 f"-e DB_PASSWORD='{ conn_foo_db.password }' "
                 f' { image }  '
                 'app=deploy ',
    dag=dag,
)

for task in tasks_list:
    task_op = BashOperator(
        task_id=f'run_{task}_task',
        bash_command='docker run '
                     '-i --rm '
                     f"-e DB_PASSWORD='{conn_foo_db.password}' "
                     f' {image}  '
                     f'app=do_{task}',
        dag=dag,
    )
    refresh_data >> task_op >> deploy

由于默认触发规则为 ALL_SUCESS，因此 deploy 仅在 tasks_list 中的所有任务都成功后才开始运行。

一些注意事项:

1. 您多次使用相同的代码，您可能需要考虑创建某种配置文件，其中包含依赖项以及设置运算符所需的所有信息，然后使用工厂方法从该文件构造运算符，从而避免重复DAG 文件中的代码。
2. 避免访问存储在运算符(operator)范围之外的 Airflow 元存储中的连接。这是一个不好的做法。 Airflow 定期扫描您的 DAG 文件(根据 min_file_process_interval )，这将导致数据库容量较高。