hadoop组件---spark实战-----airflow----调度工具airflow定时运行任务的理解

我们在前面已经初步了解了 airflow:

hadoop组件—spark实战-----airflow----调度工具airflow的介绍和使用示例

但是 我们开始 尝试使用 airflow的 定时任务的时候 ，常常遇到一个尴尬的情况， 任务没有成功运行，或者说 设置开始时间是今天，但是 明天 才开始运行。

本篇文章 尝试 说明 其中的 原理。

首先 需要声明:

schedule_interval 的crontab的语法与 linux的crontab语法 类似，但是 运行的原理并不一样，它需要与airflow中的 start_date等设置协同生效。

几个概念

我们首先来结合实例，看几个概念

以DAG流程projects为例。

案例一

现在是2020-03-12 00:00:00

我计划让一个 任务在每天凌晨两点运行。

使用配置

tz = pytz.timezone('Asia/Shanghai')

default_args = {
    'depends_on_past': False,
    'start_date': datetime(2020, 3, 12, tzinfo=tz)
}

dag = DAG(
    "projects",
    default_args=default_args,
    schedule_interval='28 2 * * *')

我们会发现 这个任务并不会在 2020-3-12的2点28分执行，而是会在 2020-3-13的2点28分第一次执行。

案例二

假设当前时间是 2020-03-12 15:05 。

我们在projects的python代码中设置如下:

default_args = {
    'depends_on_past': False,
    'start_date': airflow.utils.dates.days_ago(1)
}

dag = DAG(
    "projects",
    default_args=default_args,
    schedule_interval='28 2 * * *')

我们期望projects流程 从昨天开始 每天 凌晨2点28分执行，然后 启动scheduler。

会发现airflow成功触发了定时运行任务。 输出的时间节点如下:

Schedule:   28 2 * * *

Last Run： 2020-03-11 02:28

Start Date:  2020-03-12 15:05

则运行情况如下:

概念理解

start_date 配置中使用的变量，含义为 调度计划开始时间 是一个静态的 期望值
Start Date 区别与start_date，指的是 调度计划开始运行的 时间， 是一个动态的 实际值
schedule interval 调度执行周期间隔
execution date 执行时间，具体任务开始运行的时间， 是一个动态的 实际值
Last Run 上一次的 execution date
period 窗口周期，一段有start 和 end 的时间段

原理

官方中的说法是

Let’s Repeat That The scheduler runs your job one schedule_interval AFTER the start date, at the END of the period 。

Airflow sets execution_date based on the left bound of the schedule period it is covering, not based on when it fires (which would be the right bound of the period)

也就是说 调度计划的 执行时间 execution date 应该是 start_date + schedule interval 。

但 任务的真正 运行 时间 其实 是在 周期的 末尾end，也就是 第二个满足 schedule interval的时间点。

第一次任务真正运行的时间 是 配置的 start_date 的第二个满足 schedule interval 的时间点

并且记录的 execution date 为 配置的 start_date 的第一个满足 schedule interval 的时间点，也就是Last Run。

另外，当作业已经执行过之后，start_date 的配置将不会再生效，这个作业的调度开始时间将直接按照上次调度所对应的 execution date 来计算。

也就是 说 airflow会在start_date开始后，符合schedule_interval定义的第一个时间点记为execution_date，但是会在下个时间点到达是才开始运行任务，也就是说由于schedule interval 是一个窗口周期，它是一段的，并且在这一段时间 的end 才会开始运行， 因为这个窗口的原因，表现处理 我们的定时任务 实际执行会滞后一个周期。

怎么让任务当天运行

我们根据原理知道 之所以 任务不会当天运行 是因为 当我们的 时间间隔 以 天为单位时 ，比如 每天2点28 .

它就会在 第一个时间间隔的时间点 开始任务， 但是 实际上 是在 第二个时间 间隔的 时间点 才真正执行任务。

也就是 说 如果我想让 任务当天 运行 只要把 时间间隔 使用小时 或者 分钟来 表示就可以了 。

把schedule_interval设短一点。

比如设置成*/20 15 * * *这样的形式。

dag会在15:00-16:00间每隔20分钟跑一次

也就是 在 15点40的时候会 进行 第一次任务的真正运行

需要注意的是 防止 重复多次运行 程序需要有判断重复运行的 逻辑

airflow最佳实践

airflow官方实际上不推荐使用 动态的start_date。

We recommend against using dynamic values as start_date, especially datetime.now() as it can be quite confusing. The task is triggered once the period closes, and in theory an @hourly DAG would never get to an hour after now as now() moves along.