example_dags/cron_replacement.py

[alexanderdean]

I really like the concept of Airflow but I've been stumped trying to get a simple cron replacement example DAG working.

I took the example_bash_operator.py and changed args like so:

 args = {
    'owner': 'airflow',
    'start_date': datetime.now(),
    'schedule_interval': timedelta(minutes = 5)
}

I was expecting this DAG to start now and run every 5 minutes, but instead the scheduler runs the DAG in a loop, back-to-back.

I'd imagine migrating from cron to Airflow would be a pretty common use case - it would be awesome to see an example DAG for replacing e.g. a cron that runs every 6 hours.

[alexanderdean]

There's a little more information here: #7.

I think what might be happening is that the scheduler is kicking off from the date of the 'start_date', ignoring the hours/mins/secs, and then running every 5 minutes to "catch up" until present time.

But that would seem to be contradicted by the documentation:

start_date (datetime.datetime) – The timestamp ...

Note datetime.datetime (not datetime.date), and "timestamp" reference...

[mistercrunch]

schedule_interval currently can only be defined at the DAG level, when creating a DAG instance. If you write dag = DAG(default_args=args, schedule_interval=timedelta(minutes=5)) you'll get a 5 minutes DAG. The schedule_interval at the task level is currently disregarded and the one from the DAG level is forced. In theory you could have a DAG with tasks running at different intervals but that may be slightly confusing so I switched that feature off for now. I could re-enable it easily.

The scheduler will start on start_date, and attempt to run for every schedule_interval from that point. I think the scheduler is set to trigger every minute by default.

[alexanderdean]

Thanks for your speedy answer Maxime! If I pass in datetime.datetime(2015, 6, 15, 7, 28, 48, 163296) as my start_date, then does the scheduler start:

1. At datetime.datetime(2015, 6, 15, 7, 28, 48, 163296), or
2. At datetime.datetime(2015, 6, 15, 0, 0, 0, 0)?

(BTW, I agree - schedule_interval at the DAG and task level seems confusing, good to hear that's being removed.)

[mistercrunch]

The key for a task instance is (dag_id, task_id, execution_date). Your first execution_date is == to your start_date, then your following execution_date will be start_date + schedule_interval (and so on).

So. You should definitely round down your start_date to match your schedule_interval. Meaning if you run an hourly DAG, your start_date should be truncated to a specific hour, otherwise that's just confusing to have trailing minutes/seconds.

Also, if we were to support DAGs with heterogenous schedule_interval, (say a mix of hourly and daily), you'd want them to align on the execution_date, meaning every 24 hours you want for the hourly job to line up with your daily job, since the dependencies are checked based on the execution_date

[alexanderdean]

Many thanks @mistercrunch - your description doesn't exactly match what I saw, but it might well have been pilot error on my part. I'll give it another try and get back to you either way...

[rothnic]

I was trying this schedule_interval example and cannot get it to work with the current pypi version (1.0.1) and github version. If I pass in the schedule_interval to DAG it applies it, but as part of the default args it is not used.

[mistercrunch]

Current pypi is 1.3.1 I believe, but schedule_interval is a DAG only param for now, I could add heterogenous schedule_interval support easily but I need to put some safeguards on it first:
Related: #221

[rothnic]

I can't remember exactly but I think I was providing it to DAG, just that it only works if given explicitly, not as a default args. Or are the default args not for the DAG?

[mistercrunch]

default_args is really just for operators constructor calls. Maybe it should be called task_default_args
It's useful there because people typically create many tasks and tasks share a lot of parameters.

I clarified the docs yesterday around schedule_interval. I'd make schedule_interval a required arg for DAG, but I don't want to break existing DAGs...