气流和时间调度不工作时，运行每5分钟

Question

我不得不说气流和运行cron非常令人困惑。我只是想从现在开始每5分钟做一次cron。创建一个复杂的日期很容易。根据文档找出运行cron的逻辑不是。

当我运行下面的代码时会发生什么?每运行一秒，打印以下内容:

1 2016-08-26 17:17:01.584360
2 2016-08-26 17:17:08.124035
3 2016-08-26 17:17:15.293874
1 2016-08-26 17:17:24.100623
2 2016-08-26 17:17:31.637739
3 2016-08-26 17:17:37.919901
1 2016-08-26 17:17:45.255641
2 2016-08-26 17:17:52.859954
3 2016-08-26 17:17:59.048536
1 2016-08-26 17:18:06.175670
2 2016-08-26 17:18:12.759000
3 2016-08-26 17:18:20.112758
1 2016-08-26 17:18:26.909130
2 2016-08-26 17:18:34.396926
on and on....WOWEE

下面是我的代码。

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators import PythonOperator
from airflow.operators import TriggerDagRunOperator
from airflow.operators import DummyOperator
from datetime import datetime, timedelta
def checkfornewdata():
    f = open('/tmp/first_text.log','a')
    f.write('1 %sn' % datetime.now())
    f.close()
    return True
def fetchdata():
    f = open('/tmp/first_text.log','a')
    f.write('2 %sn' % datetime.now())
    f.close()
    return True
def uploadtoes():
    f = open('/tmp/first_text.log','a')
    f.write('3 %sn' % datetime.now())
    f.close()
    return True

mytime = datetime.combine(datetime.now()-timedelta(minutes=5),
                                  datetime.min.time())

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    "start_date": mytime,
    'email': ['[e-mail]'],
    'email_on_failure': True,
    'email_on_retry': True,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
}
# */5 * * * *
dag = DAG('first_test', schedule_interval="*/5 * * * *", default_args=default_args)
node_0 = PythonOperator(
    task_id='isnewdata',
    provide_context=False,
    python_callable=checkfornewdata,
    dag=dag)

node_0_1 = PythonOperator(
    task_id='fetchdata',
    provide_context=False,
    python_callable=fetchdata,
    dag=dag)
node_0_1_2 = PythonOperator(
    task_id='uploadtoes',
    provide_context=False,
    python_callable= uploadtoes,
    dag=dag)

node_0_1.set_upstream(node_0)
node_0_1_2.set_upstream(node_0_1)

Answer 1

至少，您希望将"start_date": mytime更改为使用非动态时间。请特别参阅FAQ。