大数据-252 离线数仓 - Airflow 任务调度 Crontab简介任务集成部署入门案例

Crontab简介

基础介绍

Linux系统是由cron（crond）系统服务来控制的，Linux系统上原本就有非常多的计划性工作，因此这个系统服务是默认启动的。 Linux系统也提供了Linux用户控制计划任务的命令：crontab命令

日志文件：ll /var/log/cron*
编辑文件: vim /etc/crontab
进程: ps -ef | grep crond => /etc/init.d/crond restart
作用：任务（命令）定时调度、定时备份等

格式说明

字段	范围	描述
第一个*	分钟（0-59）	每小时的第几分钟执行
第二个*	小时（0-23）	每天的第几个小时执行
第三个*	日期（1-31）	每月的第几天执行
第四个*	月份（1-12）	每年的第几个月执行
第五个*	星期（0-7）	每周的第几天执行

以上各个字段中，还可以使用以下特殊字符：

* 代表所有的取值范围内的数字，如月份字段为*，则表示1到12个月
/ 代表每一定时间间隔的意思，如分钟字段为*/10，表示每10分钟执行1次
- 代表从某个区间范围，是闭区间，如2-5表示2,3,4,5，小时字段中0-23/2表示在0~23点范围内每两小时执行一次
，分散的数字（不连续），如1,2,3,6,8,9
由于各个地方每周的第一天不一样，因此Sunday=0（第一天）或Sunday=7（最后一天）

配置实例

# 每一分钟执行一次command（因cron默认每1分钟扫描一次，因此全为*即可）
* * * * * command

# 每小时的第3和第15分钟执行command
3,15 * * * * command

# 每天上午8-11点的第3和第15分钟执行command
3,15 8-11 * * * command

# 每隔2天的上午8-11点的第3和第15分钟执行command
3,15 8-11 */2 * * command

# 每个星期一的上午8点到11点的第3和第15分钟执行command
3,15 8-11 * * 1 command

# 每晚的21:30执行command
30 21 * * * command

# 每月1、10、22日的4:45执行command
45 4 1,10,22 * * command

# 每周六、周日的1:10执行command
10 1 * * 6,0 command

# 每小时执行command
0 */1 * * * command

# 晚上11点到早上7点之间，每隔一小时执行command
* 23-7/1 * * * command

任务集成部署

Airflow核心概念

DAGs

有向无环图（Directed Acyclic Graph），将所有需要运行的tasks按照依赖关系组织起来，描述的是所有tasks执行的顺序

Operators

Airflow内置了很多Operators：

BashOperator 执行一个Bash命令
PythonOperator 调用任意的Python函数
EmailOperator 用于发送邮件
HTTPOperator 用于发送HTTP请求
SqlOperator 用于执行SQL命令
自定义 Operator

Task

Task是Operator的一个实例

Task Instance

由于Task会被重复调度，每次Tasks的运行就是不同的Task Instance，Task Instance 有自己的状态，包括 success、running、failed、skipped、up_for_rechedule、up_for_retry、queued、no_status等

Task Relationships

DAGs中的不同Tasks之间可以有依赖关系

Executor

在Airflow中支持的执行器有四种：

SequentialExecutor：单进程顺序执行任务，默认执行器，通常只用于测试
LocalExecutor：多进程本地执行任务
CeleryExecutor：分布式调度，生产常用，Celery是一个分布式调度框架，需要使用第三方组件，如RabbitMQ
DaskExecutor：动态任务调度，主要用于数据分析

入门案例

编写脚本

创建DAG目录并编写脚本：

mkdir $AIRFLOW_HOME/dags
vim $AIRFLOW_HOME/dags/helloworld.py

写入内容：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.utils import dates
from airflow.utils.helpers import chain
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator

# 定义默认参数
def default_options():
    default_args = {
        'owner': 'airflow',  # 拥有者名称
        'start_date': dates.days_ago(1),  # 第一次开始执行的时间
        'retries': 1,  # 失败重试次数
        'retry_delay': timedelta(seconds=5)  # 失败重试间隔
    }
    return default_args

# 定义Bash任务
def task1(dag):
    t = "pwd"
    task = BashOperator(
        task_id='MyTask1',  # task_id
        bash_command=t,  # 指定要执行的命令
        dag=dag  # 指定归属的dag
    )
    return task

# Python任务函数
def hello_world():
    current_time = str(datetime.today())
    print('hello world at {}'.format(current_time))

# 定义Python任务
def task2(dag):
    task = PythonOperator(
        task_id='MyTask2',
        python_callable=hello_world,  # 指定要执行的函数
        dag=dag
    )
    return task

# 定义另一个Bash任务
def task3(dag):
    t = "date"
    task = BashOperator(
        task_id='MyTask3',
        bash_command=t,
        dag=dag
    )
    return task

# 定义DAG
with DAG(
    'HelloWorldDag',  # dag_id
    default_args=default_options(),  # 指定默认参数
    schedule_interval="*/2 * * * *"  # 执行周期，每分钟2次
) as d:
    task1 = task1(d)
    task2 = task2(d)
    task3 = task3(d)
    chain(task1, task2, task3)  # 指定执行顺序

测试运行

# 执行命令检查脚本是否有错误
python $AIRFLOW_HOME/dags/helloworld.py

# 查看生效的dags
airflow dags list --subdir $AIRFLOW_HOME/dags

# 查看指定dag中的task
airflow tasks list HelloWorldDag

# 测试dag中的task
airflow tasks test HelloWorldDag MyTask2 2020-08-01

大数据-252 离线数仓 - Airflow 任务调度 Crontab简介 任务集成部署 入门案例