当前位置：首页 > news >正文

Apache Airflow (八) ：DAG任务依赖设置

news 来源：原创 2024/5/20 12:17:25

🏡 个人主页：IT贫道_大数据OLAP体系技术栈,Apache Doris,Clickhouse 技术-CSDN博客

🚩 私聊博主：加入大数据技术讨论群聊，获取更多大数据资料。

🔔 博主个人B栈地址：豹哥教你大数据的个人空间-豹哥教你大数据个人主页-哔哩哔哩视频

1. DAG任务依赖设置一

2. DAG任务依赖设置二

3. DAG任务依赖设置三

4. DAG任务依赖设置四

5. DAG任务依赖设置五

1. DAG任务依赖设置一

DAG调度流程图

task执行依赖

A >> B >>C

完整代码

'''
airflow 任务依赖关系设置一'''
from airflow import DAGfrom airflow.operators.bash import BashOperatorfrom datetime import datetime, timedelta
default_args = {'owner': 'airflow', # 拥有者名称'start_date': datetime(2021, 9, 22),  # 第一次开始执行的时间，为 UTC 时间'retries': 1,  # 失败重试次数'retry_delay': timedelta(minutes=5),  # 失败重试间隔
}dag = DAG(dag_id = 'dag_relation_1', #DAG id ,必须完全由字母、数字、下划线组成default_args = default_args, #外部定义的 dic 格式的参数schedule_interval = timedelta(minutes=1) # 定义DAG运行的频率,可以配置天、周、小时、分钟、秒、毫秒
)A = BashOperator(task_id='A',bash_command='echo "run A task"',dag=dag
)B = BashOperator(task_id='B',bash_command='echo "run B task"',dag=dag
)C = BashOperator(task_id='C',bash_command='echo "run C task"',dag=dag,retries=3
)A >> B >>C

2. DAG任务依赖设置二

DAG调度流程图

task执行依赖

[A,B] >>C >>D

完整代码

'''
airflow 任务依赖关系设置二'''
from airflow import DAGfrom airflow.operators.bash import BashOperatorfrom datetime import datetime, timedelta
default_args = {'owner': 'airflow', # 拥有者名称'start_date': datetime(2021, 9, 22),  # 第一次开始执行的时间，为 UTC 时间'retries': 1,  # 失败重试次数'retry_delay': timedelta(minutes=5),  # 失败重试间隔
}dag = DAG(dag_id = 'dag_relation_2', #DAG id ,必须完全由字母、数字、下划线组成default_args = default_args, #外部定义的 dic 格式的参数schedule_interval = timedelta(minutes=1) # 定义DAG运行的频率,可以配置天、周、小时、分钟、秒、毫秒
)A = BashOperator(task_id='A',bash_command='echo "run A task"',dag=dag
)B = BashOperator(task_id='B',bash_command='echo "run B task"',dag=dag
)C = BashOperator(task_id='C',bash_command='echo "run C task"',dag=dag,retries=3
)D = BashOperator(task_id='D',bash_command='echo "run D task"',dag=dag
)[A,B] >>C >>D

3. DAG任务依赖设置三

DAG调度流程图

task执行依赖

[A,B,C] >>D >>[E,F]

完整代码

'''
airflow 任务依赖关系设置三'''
from airflow import DAGfrom airflow.operators.bash import BashOperatorfrom datetime import datetime, timedelta
default_args = {'owner': 'airflow', # 拥有者名称'start_date': datetime(2021, 9, 22),  # 第一次开始执行的时间，为 UTC 时间'retries': 1,  # 失败重试次数'retry_delay': timedelta(minutes=5),  # 失败重试间隔
}dag = DAG(dag_id = 'dag_relation_3', #DAG id ,必须完全由字母、数字、下划线组成default_args = default_args, #外部定义的 dic 格式的参数schedule_interval = timedelta(minutes=1) # 定义DAG运行的频率,可以配置天、周、小时、分钟、秒、毫秒
)A = BashOperator(task_id='A',bash_command='echo "run A task"',dag=dag
)B = BashOperator(task_id='B',bash_command='echo "run B task"',dag=dag
)C = BashOperator(task_id='C',bash_command='echo "run C task"',dag=dag,retries=3
)D = BashOperator(task_id='D',bash_command='echo "run D task"',dag=dag
)E = BashOperator(task_id='E',bash_command='echo "run E task"',dag=dag
)F = BashOperator(task_id='F',bash_command='echo "run F task"',dag=dag
)[A,B,C] >>D >>[E,F]

4. DAG任务依赖设置四

DAG调度流程图

task执行依赖

A >>B>>C>>D
A >>E>>F

完整代码

'''
airflow 任务依赖关系设置四'''
from airflow import DAGfrom airflow.operators.bash import BashOperatorfrom datetime import datetime, timedelta
default_args = {'owner': 'airflow', # 拥有者名称'start_date': datetime(2021, 9, 22),  # 第一次开始执行的时间，为 UTC 时间'retries': 1,  # 失败重试次数'retry_delay': timedelta(minutes=5),  # 失败重试间隔
}dag = DAG(dag_id = 'dag_relation_4', #DAG id ,必须完全由字母、数字、下划线组成default_args = default_args, #外部定义的 dic 格式的参数schedule_interval = timedelta(minutes=1) # 定义DAG运行的频率,可以配置天、周、小时、分钟、秒、毫秒
)A = BashOperator(task_id='A',bash_command='echo "run A task"',dag=dag
)B = BashOperator(task_id='B',bash_command='echo "run B task"',dag=dag
)C = BashOperator(task_id='C',bash_command='echo "run C task"',dag=dag,retries=3
)D = BashOperator(task_id='D',bash_command='echo "run D task"',dag=dag
)E = BashOperator(task_id='E',bash_command='echo "run E task"',dag=dag
)F = BashOperator(task_id='F',bash_command='echo "run F task"',dag=dag
)A >>[B,C,D]
A >>[E,F]

5. DAG任务依赖设置五

DAG调度流程图

task执行依赖

A >>B>>E
C >>D>>E

完整代码

'''
airflow 任务依赖关系设置五'''
from airflow import DAGfrom airflow.operators.bash import BashOperatorfrom datetime import datetime, timedelta
default_args = {'owner': 'airflow', # 拥有者名称'start_date': datetime(2021, 9, 22),  # 第一次开始执行的时间，为 UTC 时间'retries': 1,  # 失败重试次数'retry_delay': timedelta(minutes=5),  # 失败重试间隔
}dag = DAG(dag_id = 'dag_relation_5', #DAG id ,必须完全由字母、数字、下划线组成default_args = default_args, #外部定义的 dic 格式的参数schedule_interval = timedelta(minutes=1) # 定义DAG运行的频率,可以配置天、周、小时、分钟、秒、毫秒
)A = BashOperator(task_id='A',bash_command='echo "run A task"',dag=dag
)B = BashOperator(task_id='B',bash_command='echo "run B task"',dag=dag
)C = BashOperator(task_id='C',bash_command='echo "run C task"',dag=dag,retries=3
)D = BashOperator(task_id='D',bash_command='echo "run D task"',dag=dag
)E = BashOperator(task_id='E',bash_command='echo "run E task"',dag=dag
)A >>B>>E
C >>D>>E