在一个DAG内执行顺序和并发的任务

Question

我是新来的气流，有一些关于如何正确运行一些任务的基本问题并发和其他依次

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

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

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

我只希望在[this, that, the_other_thing]完成后部署，但不清楚三个中哪个将最后完成。在一个DAG内执行此序列的最佳实践是什么?设置concurrency=3并在for循环中执行[this, that, the_other_thing]是否足够?欢迎提出任何建议

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，所以只有在tasks_list中的所有任务都成功后，deploy才会开始运行。

注释:

1. 如果您多次使用相同的代码，您可能需要考虑创建某种类型的配置文件，其中包含设置操作符所需的依赖关系和所有信息，然后使用工厂方法从该文件构造操作符，从而避免在DAG文件中重复代码。
2. 避免访问操作员范围之外存储在气流介孔中的连接。这是一种不好的做法。气流定期扫描您的DAG文件(根据min_file_process_interval)，这会导致DB上的高音量。