Apache Airflow中调度程序间隔和启动时间的工作不正确

Question

任务开始时间找不到解决方案。我有代码，找不到我错了。

当我运行DAG时，分别为25.03、26.03、27.03。任务已经完成，但是今天（28.03）的任务没有在6:48开始。

我尝试使用cron表达式，钟摆，日期时间和结果相同。当地时间（UTC + 3）和气流时间（UTC）不同。我尝试在“开始日期”或“时间表间隔”中使用每次（本地，空气流量），但没有结果。

使用：Ubuntu，Airflow 1.9.0版和本地执行程序。

emailname = Variable.get('test_mail')
l_start_date = datetime(2018, 3, 25, 6, 48) 
l_schedule_interval = '@daily'

WORKFLOW_DEFAULT_ARGS = {
    'owner': 'owner',
    'depends_on_past': True,
    'start_date': l_start_date,
    'email': emailname,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retries_delay': timedelta(minutes=1),
}

# initialize the DAG
dag = DAG(
    dag_id='test_dag_mail',
    default_args=WORKFLOW_DEFAULT_ARGS,
    schedule_interval=l_schedule_interval,
    start_date=l_start_date,
 )

Answer 1

该调度页面做解释，该方面做得很差execution_date设定为在上次运行和当前运行的周期的开始。为什么？好吧，假设每月，每天，每小时的工作需要从该时间开始读取所有数据，直到当前间隔时间为止。是的，他们本来可以改变它的。

示例（将纳秒/毫秒转换为个位数）：

DAG start_date  DAG schedule_interval Task Started          Task execution_date
2017-12-01      '@daily'              2017-12-02 00:00:02.8 2017-12-01 00:00:00.0
                                      2017-12-03 00:00:01.4 2017-12-02 00:00:00.0
2017-12-01      '@weekly'             2017-12-08 00:00:01.5 2017-12-01 00:00:00.0
                                      2017-12-15 00:00:03.9 2017-12-08 00:00:00.0
2017-12-01      '33 03 * * *'         2017-12-02 03:33:01.6 2017-12-01 03:33:00.0
                                      2017-12-03 03:33:02.2 2017-12-02 03:33:00.0
2017-12-01      '33 03 * * 2'         2017-12-12 03:33:01.7 2017-12-05 03:33:00.0
                                      2017-12-19 03:33:03.1 2017-12-12 03:33:00.0

对于后一种情况，给定节点，开始日期不是间隔中所要求的星期二；因为开始时间可能与间隔时间不符，这令人困惑，所以我尚未对此进行精确测试，建议您如果要在每个星期二运行，则开始时间应该在星期二，例如2017-12-05或2017- 11-28。

2017-12年的参考：

#  December 2017
Su Mo Tu We Th Fr Sa
                1  2
 3  4  5  6  7  8  9
10 11 12 13 14 15 16
17 18 19 20 21 22 23
24 25 26 27 28 29 30
31

因此，在DAG的任务中，使用提供的运行上下文中的jinja2模板使用{{ds}}或{{execution_date}}将引用示例的最后一列，而不是datetime.now()告诉您第二到最后一列的内容。

提供这些信息是为了使您的任务是幂等的：f（data）= f（f（data））。

如果您运行DAG一次，则最终状态应与您运行N次DAG相同。这样，如果您运行（线性）DAG，并且5个任务中的3个成功完成，但是任务4失败并且任务5从未运行，则可以重新运行整个dag，任务1-3将预先清除或覆盖数据，以便这些任务的先前成功不会改变输出，然后任务4和5有望成功，您将处于最终的良好状态。