大数据调度平台Airflow（五）：Airflow使用

Airflow使用

一、Airflow调度Shell命令

1.首先我们需要创建一个python文件，导入需要的类库

2.实例化DAG

3、定义Task

4、设置task依赖关系

5、上传python配置脚本

6、重启Airflow

7、执行airflow

二、DAG调度触发时间

三、DAG catchup 参数设置

四、DAG调度周期设置

五、DAG任务依赖设置

1、DAG任务依赖设置一

2、DAG任务依赖设置二

3、DAG任务依赖设置三

4、DAG任务依赖设置四

5、DAG任务依赖设置五

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:ProgramDataAnaconda3envspython37scripts>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博客大数据系列文章会每天更新，停下休息的时候不要忘了别人还在奔跑，希望大家抓紧时间学习，全力奔赴更美好的生活✨

大数据调度平台Airflow（五）：Airflow使用

一、Airflow调度Shell命令 下面我们以调度执行shell命令为例，来讲解Airflow使用。

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调度Shell命令
下面我们以调度执行shell命令为例，来讲解Airflow使用。

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任务依赖设置