1

我不得不说气流和运行 cron 非常令人困惑。我只想从现在开始每 5 分钟启动一次 cron。创建一个复杂的 dag 很容易。不是根据文档弄清楚运行 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 %s\n' % datetime.now())
    f.close()
    return True

def fetchdata():
    f = open('/tmp/first_text.log','a')
    f.write('2 %s\n' % datetime.now())
    f.close()
    return True

def uploadtoes():
    f = open('/tmp/first_text.log','a')
    f.write('3 %s\n' % 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': ['test@gmail.com'],
    '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)
4

1 回答 1

1

至少，您需要更改"start_date": mytime为使用非动态时间。请特别参阅常见问题解答。

于 2017-01-20T12:03:35.943 回答