CODEWITHSUNDEEP
pythonairflow
Tampa
Tampa

Reputation: 78362

airflow and cron - scheduling does not work when running ever 5 mins

I have to say airflow and running a cron is very confusing. I just want to start a cron every 5 mins from NOW. Creating a complex dag is easy. Figuring out the logic of running a cron based on the docs is not.

What happens when I run the below code? Ever second it runs print the below:

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

Below is my code.

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': ['[email protected]'],
    '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)

Upvotes: 1

Views: 2160

Answers (1)

russellpierce
russellpierce

Reputation: 4711

At the very least, you'll want to change "start_date": mytime to use a non-dynamic time. See particularly the FAQ.

Upvotes: 2

Related Questions