气流DAG触发

Question

我们最近尝试采用气流作为我们的“数据工作流”引擎，虽然我已经完成了大部分工作，但我仍然处在调度器如何计算何时触发DAG的灰色区域。

看看这个简单的dag：

from airflow import DAG
from datetime import datetime
from airflow.operators.bash_operator import BashOperator

dag_options = {                
            'owner':                'Airflow',  
            'depends_on_past':      False,      
            'start_date':           datetime.now()
}

with DAG('test_dag1', schedule_interval="5 * * * *", default_args=dag_options) as dag:
                task1 = BashOperator(      
                task_id='task1', 
                bash_command='date',                
                dag=dag)      

计划会处理这个问题，但是不会执行它。现在，如果我将"start_date“改为：

datetime(year=xxxx,month=yyyy=day=zzzz) 

如果xxxx，yyyy，zzzz是今天的日期，它将开始执行。是因为调度程序不断从源dag文件夹重新读取这个dag，每次执行datetime.now()，注意到开始日期与当前排队的不同，重新添加这个守护进程，从而重新调度/推迟执行日期(我的dag_dir_list_interval是300)吗？

此外，在气流中，据我所知，当进程未暂停(或添加dags_are_paused_at_creation = False)时，调度程序将按以下方式调度执行：

• 第一批执行:即时后(start_date + interval)
• 第二批执行:立即执行(start_date +(间隔* 2))
• 第三批执行:立即执行(start_date +(间隔* 3))

这是正确的假设吗？

更新(7/30/2017)

基于上述假设，我今天创建了这个数据库(07/30/2017)：

from airflow import DAG
from datetime import datetime
from airflow.operators.bash_operator import BashOperator

dag_options = {                
            'owner':             'Airflow',  
            'depends_on_past':   False,      
            'start_date':   
datetime(year=2017,month=7,day=30,hour=20,minute=10)
}

with DAG('test_dag_100', schedule_interval="*/10 * * * *", 
default_args=dag_options) as dag:
                task1 = BashOperator(      
                task_id='task_100', 
                bash_command='date',                
                dag=dag)      

应从(世界协调时)开始：

• 2017年7月30日20:20:00
• 2017年7月30日20:30
• 2017年7月30日20:40:00

不幸的是，这种情况并没有发生。下面是我的仪表板的一些屏幕截图：

• dag的详细信息
• 第一次执行应该是在什么时候

有人能解释一下为什么在20:21的时候达格没有执行吗？20点31分之后，它仍然没有执行.我在这里错过了什么？

顺便说一句，我还注意到，由于某种原因，每次我通过仪表板手动启动一个达格时，它都处于“运行”阶段。为什么会这样呢？手动启动是否与任何启动时间选项( start _date/interval/等等)有关?？

谢谢你所能提供的任何澄清

Answer 1

你的假设是正确的。气流将安排第一次DAG运行后，指定的时间表间隔已经过去，从开始日期。使用datetime.now()作为启动日期将导致很少(如果有可能)触发DAG的气流。调度文档中提到了这一点。

如果要指定特定的开始日期，例如日期时间(207，7，27，0)，计划间隔为"5 ** *"，那么在7/27的凌晨1:05分，DAG将被触发以第一次运行。在那之后，它将继续每五分钟运行一次。