大数据调度平台Airflow（五）：Airflow使用
目录
Airflow使用
上文说到使用Airflow进行任务调度大体步骤如下：
- 创建python文件，根据实际需要，使用不同的Operator
- 在python文件不同的Operator中传入具体参数，定义一系列task
- 在python文件中定义Task之间的关系，形成DAG
- 将python文件上传执行，调度DAG，每个task会形成一个Instance
- 使用命令行或者WEBUI进行查看和管理
以上python文件就是Airflow python脚本，使用代码方式指定DAG的结构
一、Airflow调度Shell命令
下面我们以调度执行shell命令为例，来讲解Airflow使用。
1.首先我们需要创建一个python文件，导入需要的类库
-
# 导入 DAG 对象，后面需要实例化DAG对象
-
from airflow import DAG
-
-
# 导入BashOperator Operators,我们需要利用这个对象去执行流程
-
from airflow.operators.bash import BashOperator
注意：以上代码可以在开发工具中创建，但是需要在使用的python3.7环境中导入安装Airflow包。
D:\ProgramData\Anaconda3\envs\python37\Scripts>pip install apache-airflow==2.1.3 -i https://pypi.tuna.tsinghua.edu.cn/simple
2.实例化DAG
-
from datetime import datetime, timedelta
-
-
# default_args中定义一些参数，在实例化DAG时可以使用，使用python dic 格式定义
-
default_args = {
-
'owner': 'airflow', # 拥有者名称
-
'start_date': datetime(2022, 3, 25), # 第一次开始执行的时间，为 UTC 时间
-
'retries': 1, # 失败重试次数
-
'retry_delay': timedelta(minutes=5), # 失败重试间隔
-
}
-
-
dag = DAG(
-
dag_id = 'myairflow_execute_bash', #DAG id ,必须完全由字母、数字、下划线组成
-
default_args = default_args, #外部定义的 dic 格式的参数
-
schedule_interval = timedelta(days=1) # 定义DAG运行的频率,可以配置天、周、小时、分钟、秒、毫秒
-
)
注意：
- 实例化DAG有三种方式
第一种方式：
-
with DAG("my_dag_name") as dag:
-
op=XXOperator(task_id="task")
第二种方式（以上采用这种方式）：
-
my_dag = DAG("my_dag_name")
-
op = XXOperator(task_id="task", dag=my_dag)
第三种方式：
-
@dag(start_date=days_ago(2))
-
def generate_dag():
-
op = XXOperator(task_id="task")
-
dag = generate_dag()
- baseoperator基础参数说明：
可以参照：
http://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/models/baseoperator/index.html#module-airflow.models.baseoperator查看baseopartor中更多参数。
- DAG参数说明
可以参照：
http://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/models/dag/index.html
查看DAG参数说明，也可以直接在开发工具点击DAG进入源码看下对应参数有哪些。
3、定义Task
当实例化Operator时会生成Task任务，从一个Operator中实例化出来对象的过程被称为一个构造方法，每个构造方法中都有“task_id”充当任务的唯一标识符。
下面我们定义三个Operator，也就是三个Task，每个task_id 不能重复。
-
# operator 支持多种类型， 这里使用 BashOperator
-
first = BashOperator(
-
task_id='first',
-
bash_command='echo "run first task"',
-
dag=dag
-
)
-
-
middle = BashOperator(
-
task_id='middle',
-
bash_command='echo "run middle task"',
-
dag=dag
-
)
-
-
last = BashOperator(
-
task_id='last',
-
bash_command='echo "run last task"',
-
dag=dag,
-
retries=3
-
)
注意：
- 每个operator中可以传入对应的参数，覆盖DAG默认的参数，例如：last task中“retries”=3 就替代了默认的1。任务参数的优先规则如下：①.显示传递的参数 ②.default_args字典中存在的值③.operator的默认值（如果存在）。
- BashOperator使用方式参照：http://airflow.apache.org/docs/apache-airflow/stable/howto/operator/bash.html#howto-operator-bashoperator
4、设置task依赖关系
-
#使用 set_upstream、set_downstream 设置依赖关系,不能出现环形链路，否则报错
-
# middle.set_upstream(first) # middle会在first执行完成之后执行
-
# last.set_upstream(middle) # last 会在 middle执行完成之后执行
-
-
#也可以使用位移符来设置依赖关系
-
first >> middle >>last # first 首先执行，middle次之，last最后
-
# first >> [middle,last] # first首先执行，middle ,last并行执行
注意：当执行脚本时，如果在DAG中找到一条环形链路（例如：A->B->C-A）会引发异常。更多DAG task依赖关系可参照官网：http://airflow.apache.org/docs/apache-airflow/stable/concepts/dags.html#task-dependencies
5、上传python配置脚本
到目前为止，python配置如下：
-
# 导入 DAG 对象，后面需要实例化DAG对象
-
from airflow import DAG
-
-
# 导入BashOperator Operators,我们需要利用这个对象去执行流程
-
from airflow.example_dags.example_bash_operator import dag
-
-
from airflow.operators.bash import BashOperator
-
-
from datetime import datetime, timedelta
-
-
# default_args中定义一些参数，在实例化DAG时可以使用，使用python dic 格式定义
-
default_args = {
-
'owner': 'airflow', # 拥有者名称
-
'start_date': datetime(2021, 9, 4), # 第一次开始执行的时间，为 UTC 时间
-
'retries': 1, # 失败重试次数
-
'retry_delay': timedelta(minutes=5), # 失败重试间隔
-
}
-
-
dag = DAG(
-
dag_id = 'myairflow_execute_bash', #DAG id ,必须完全由字母、数字、下划线组成
-
default_args = default_args, #外部定义的 dic 格式的参数
-
schedule_interval = timedelta(days=1) # 定义DAG运行的频率,可以配置天、周、小时、分钟、秒、毫秒
-
)
-
-
# operator 支持多种类型， 这里使用 BashOperator
-
first = BashOperator(
-
task_id='first',
-
bash_command='echo "run first task"',
-
dag=dag
-
)
-
-
middle = BashOperator(
-
task_id='middle',
-
bash_command='echo "run middle task"',
-
dag=dag
-
)
-
-
last = BashOperator(
-
task_id='last',
-
bash_command='echo "run last task"',
-
dag=dag,
-
retries=3
-
)
-
-
#使用 set_upstream、set_downstream 设置依赖关系,不能出现环形链路，否则报错
-
# middle.set_upstream(first) # middle会在first执行完成之后执行
-
# last.set_upstream(middle) # last 会在 middle执行完成之后执行
-
-
#也可以使用位移符来设置依赖关系
-
first >> middle >>last # first 首先执行，middle次之，last最后
-
# first >> [middle,last] # first首先执行，middle ,last并行执行
将以上python配置文件上传到$AIRFLOW_HOME/dags目录下，默认$AIRFLOW_HOME为安装节点的“/root/airflow”目录，当前目录下的dags目录需要手动创建。
6、重启Airflow
“ps aux|grep webserver”和“ps aux|grep scheduler”找到对应的airflow进程杀掉，重新启动Airflow。重启之后，可以在airflow webui看到对应的DAG ID ”myairflow_execute_bash”。
7、执行airflow
按照如下步骤执行DAG，首先打开工作流，然后“Trigger DAG”执行，随后可以看到任务执行成功。
查看task执行日志：
二、DAG调度触发时间
在Airflow中，调度程序会根据DAG文件中指定的“start_date”和“schedule_interval”来运行DAG。特别需要注意的是Airflow计划程序在计划时间段的末尾触发执行DAG，而不是在开始时刻触发DAG，例如：
-
default_args = {
-
'owner': 'airflow', # 拥有者名称
-
'start_date': datetime(2022, 3, 25), # 第一次开始执行的时间，为 UTC 时间
-
'retries': 1, # 失败重试次数
-
'retry_delay': timedelta(minutes=5), # 失败重试间隔
-
}
-
-
dag = DAG(
-
dag_id = 'myairflow_execute_bash', #DAG id ,必须完全由字母、数字、下划线组成
-
default_args = default_args, #外部定义的 dic 格式的参数
-
schedule_interval = timedelta(days=1) # 定义DAG运行的频率,可以配置天、周、小时、分钟、秒、毫秒
-
)
以上配置的DAG是从世界标准时间2022年3月24号开始调度，每隔1天执行一次，这个DAG的具体运行时间如下图：
| 自动调度DAG 执行日期 |
自动调度DAG实际执行触发时间 |
| 2022-03-24，00:00:00+00:00 |
2022-03-25，00:00:00+00:00 |
| 2022-03-25，00:00:00+00:00 |
2022-03-26，00:00:00+00:00 |
| 2022-03-26，00:00:00+00:00 |
2022-03-27，00:00:00+00:00 |
| 2022-03-27，00:00:00+00:00 |
2022-03-28，00:00:00+00:00 |
| 2022-03-28，00:00:00+00:00 |
2022-03-29，00:00:00+00:00 |
| ... ... |
... ... |
以上表格中以第一条数据为例解释，Airflow正常调度是每天00:00:00 ，假设当天日期为2022-03-24，正常我们认为只要时间到了2022-03-24 00:00:00 就会执行，改调度时间所处于的调度周期为2022-03-24 00:00:00 ~ 2022-03-25 00:00:00 ，在Airflow中实际上是在调度周期末端触发执行，也就是说2022-03-24 00:00:00 自动触发执行时刻为 2022-03-25 00:00:00。
如下图，在airflow中，“execution_date”不是实际运行时间，而是其计划周期的开始时间戳。例如：execution_date 是2021-09-04 00:00:00 的DAG 自动调度运行的实际时间为2021-09-05 00:00:00。当然除了自动调度外，我们还可以手动触发执行DAG执行，要判断DAG运行时计划调度（自动调度）还是手动触发，可以查看“Run Type”。
三、DAG catchup 参数设置
在Airflow的工作计划中，一个重要的概念就是catchup（追赶），在实现DAG具体逻辑后，如果将catchup设置为True（默认就为True）,Airflow将“回填”所有过去的DAG run，如果将catchup设置为False,Airflow将从最新的DAG run时刻前一时刻开始执行 DAG run，忽略之前所有的记录。
例如：现在某个DAG每隔1分钟执行一次，调度开始时间为2001-01-01 ，当前日期为2021-10-01 15:23:21，如果catchup设置为True，那么DAG将从2001-01-01 00:00:00 开始每分钟都会运行当前DAG。如果catchup 设置为False，那么DAG将从2021-10-01 15:22:20（当前2021-10-01 15:23:21前一时刻）开始执行DAG run。
举例：有first ,second,third三个shell命令任务，按照顺序调度，每隔1分钟执行一次，首次执行时间为2000-01-01。
设置catchup 为True（默认），DAG python配置如下：
-
from airflow import DAG
-
-
from airflow.operators.bash import BashOperator
-
-
from datetime import datetime, timedelta
-
-
default_args = {
-
'owner': 'airflow', # 拥有者名称
-
'start_date': datetime(2001, 1, 1), # 第一次开始执行的时间，为 UTC 时间
-
'retries': 1, # 失败重试次数
-
'retry_delay': timedelta(minutes=5), # 失败重试间隔
-
}
-
dag = DAG(
-
dag_id = 'catchup_test1 ', #DAG id ,必须完全由字母、数字、下划线组成
-
default_args = default_args, #外部定义的 dic 格式的参数
-
schedule_interval = timedelta(minutes=1), # 定义DAG运行的频率,可以配置天、周、小时、分钟、秒、毫秒
-
catchup=True # 执行DAG时，将开始时间到目前所有该执行的任务都执行，默认为True
-
)
-
-
first = BashOperator(
-
task_id='first',
-
bash_command='echo "run first task"',
-
dag=dag
-
)
-
middle = BashOperator(
-
task_id='second',
-
bash_command='echo "run second task"',
-
dag=dag
-
)
-
last = BashOperator(
-
task_id='third',
-
bash_command='echo "run third task"',
-
dag=dag,
-
retries=3
-
)
-
first >> middle >>last
上传python配置文件到$AIRFLOW_HOME/dags下，重启airflow,DAG执行调度如下：
设置catchup 为False,DAG python配置如下：
-
from airflow import DAG
-
-
from airflow.operators.bash import BashOperator
-
-
from datetime import datetime, timedelta
-
-
default_args = {
-
'owner': 'airflow', # 拥有者名称
-
'start_date': datetime(2001, 1, 1), # 第一次开始执行的时间，为 UTC 时间
-
'retries': 1, # 失败重试次数
-
'retry_delay': timedelta(minutes=5), # 失败重试间隔
-
}
-
dag = DAG(
-
dag_id = 'catchup_test2', #DAG id ,必须完全由字母、数字、下划线组成
-
default_args = default_args, #外部定义的 dic 格式的参数
-
schedule_interval = timedelta(minutes=1), # 定义DAG运行的频率,可以配置天、周、小时、分钟、秒、毫秒
-
catchup=False # 执行DAG时，将开始时间到目前所有该执行的任务都执行，默认为True
-
)
-
-
first = BashOperator(
-
task_id='first',
-
bash_command='echo "run first task"',
-
dag=dag
-
)
-
middle = BashOperator(
-
task_id='second',
-
bash_command='echo "run second task"',
-
dag=dag
-
)
-
last = BashOperator(
-
task_id='third',
-
bash_command='echo "run third task"',
-
dag=dag,
-
retries=3
-
)
-
first >> middle >>last
上传python配置文件到$AIRFLOW_HOME/dags下，重启airflow,DAG执行调度如下：
有两种方式在Airflow中配置catchup:
- 全局配置
在airflow配置文件airflow.cfg的scheduler部分下，设置catchup_by_default=True（默认）或False，这个设置是全局性的设置。
- DAG文件配置
在python代码配置中设置DAG对象的参数：dag.catchup=True或False。
-
dag = DAG(
-
dag_id = 'myairflow_execute_bash',
-
default_args = default_args,
-
catchup=False,
-
schedule_interval = timedelta(days=1))
四、DAG调度周期设置
每个DAG可以有或者没有调度执行周期，如果有调度周期，我们可以在python代码DAG配置中设置“schedule_interval”参数来指定调度DAG周期，可以通过以下三种方式来设置。
- 预置的Cron调度
Airflow预置了一些Cron调度周期，可以参照：
DAG Runs — Airflow Documentation，如下图：
在python配置文件中使用如下：
-
default_args = {
-
'owner': 'airflow', # 拥有者名称
-
'start_date': datetime(2021, 9, 4), # 第一次开始执行的时间，为 UTC 时间
-
'retries': 1, # 失败重试次数
-
'retry_delay': timedelta(minutes=5), # 失败重试间隔
-
}
-
-
dag = DAG(
-
dag_id = 'cron_test', #DAG id ,必须完全由字母、数字、下划线组成
-
default_args = default_args, #外部定义的 dic 格式的参数
-
schedule_interval = '@daily' # 使用预置的Cron调度,每天0点0分调度
- Cron
这种方式就是写Linux系统的crontab定时任务命令，可以在https://crontab.guru/网站先生成对应的定时调度命令，其格式如下：
-
minute hour day month week
-
minute:表示分钟，可以从0~59之间的任意整数。
-
hour:表示小时，可以是从0到23之间的任意整数。
-
day:表示日期，可以是1到31之间的任何整数。
-
month:表示月份，可以是从1到12之间的任何整数。
-
week:表示星期几，可以是从0到7之间的任何整数，这里的0或7代表星期日。
以上各个字段中还可以使用特殊符号代表不同意思：
-
星号（*）：代表所有可能的值，例如month字段如果是星号，则表示在满足其它字段的制约条件后每月都执行该命令操作。
-
逗号（,）：可以用逗号隔开的值指定一个列表范围，例如，”1,2,5,7,8,9”
-
中杠（-）：可以用整数之间的中杠表示一个整数范围，例如”2-6”表示”2,3,4,5,6”
-
正斜线（/）：可以用正斜线指定时间的间隔频率,步长，例如”0-23/2”表示每两小时执行一次。
在python配置文件中使用如下：
-
default_args = {
-
'owner': 'airflow', # 拥有者名称
-
'start_date': datetime(2021, 9, 4), # 第一次开始执行的时间，为 UTC 时间
-
'retries': 1, # 失败重试次数
-
'retry_delay': timedelta(minutes=5), # 失败重试间隔
-
}
-
-
dag = DAG(
-
dag_id = 'cron_test', #DAG id ,必须完全由字母、数字、下划线组成
-
default_args = default_args, #外部定义的 dic 格式的参数
-
schedule_interval = '* * * * *' # 使用Crontab 定时任务命令，每分钟运行一次
-
)
- datetime.timedelta
timedelta是使用python timedelta 设置调度周期，可以配置天、周、小时、分钟、秒、毫秒。在python配置文件中使用如下：
-
default_args = {
-
'owner': 'airflow', # 拥有者名称
-
'start_date': datetime(2021, 9, 4), # 第一次开始执行的时间，为 UTC 时间
-
'retries': 1, # 失败重试次数
-
'retry_delay': timedelta(minutes=5), # 失败重试间隔
-
}
-
-
dag = DAG(
-
dag_id = 'cron_test', #DAG id ,必须完全由字母、数字、下划线组成
-
default_args = default_args, #外部定义的 dic 格式的参数
-
schedule_interval = timedelta(minutes=5) # 使用python timedelta 设置调度周期，可以配置天、周、小时、分钟、秒、毫秒
-
)
五、DAG任务依赖设置
1、DAG任务依赖设置一
- DAG调度流程图
- task执行依赖
A >> B >>C
- 完整代码
-
'''
-
airflow 任务依赖关系设置一
-
-
'''
-
from airflow import DAG
-
-
from airflow.operators.bash import BashOperator
-
-
from 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 DAG
-
-
from airflow.operators.bash import BashOperator
-
-
from 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 DAG
-
-
from airflow.operators.bash import BashOperator
-
-
from 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 DAG
-
-
from airflow.operators.bash import BashOperator
-
-
from 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 DAG
-
-
from airflow.operators.bash import BashOperator
-
-
from 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
- 📢博客主页：https://lansonli.blog.csdn.net
- 📢欢迎点赞 👍 收藏 ⭐留言 📝 如有错误敬请指正！
- 📢本文由 Lansonli 原创，首发于 CSDN博客🙉
- 📢大数据系列文章会每天更新，停下休息的时候不要忘了别人还在奔跑，希望大家抓紧时间学习，全力奔赴更美好的生活✨
文章来源: lansonli.blog.csdn.net，作者：Lansonli，版权归原作者所有，如需转载，请联系作者。
原文链接：lansonli.blog.csdn.net/article/details/123746605
- 点赞
- 收藏
- 关注作者
评论（0）