Add only_run_latest to BaseOperator

[mistercrunch]

Dependencies on an only_run_latest task are satisfied when that task has succeeded at any point in the future of the task depending on it.

When backfilling, only the end_date of only_run_latest tasks would be triggered, and dependencies for this task through time would be satisfied upon completion of this latest task instance.

When it comes to an only_run_latest task, the scheduler should skip runs to max(execution_date) of all of its dependencies.

[thibault-ketterer]

+1, furthermore, to me, having "only_run_latest" feature makes the "skip/do not run" #262 useless

[martingrayson]

I have a weekly job that bulk loads some data. My understanding of the scheduler is that if I have a schedule interval of a week, once that week has elapsed and the scheduler kicks off the dag, I'll see 7 sequential executions (if depends_on_past=True).

This isn't the functionality I need - I'd like one dag run per week and dont even use the execution date parameter in my dag. Is this a solid case for only_run_latest or have I misunderstood the function? Is there a way I can achieve this prior to the feature being implemented using some sort of hack within my dag?

[mistercrunch]

I think it's unrelated, it sounds to me like you need a DAG with a schedule_interval=timedelta(7). The tasks in there will get executed once a week.

Only run latest would mean that if task a depends on task b, and task b fails for 10 days an recover after, task a will only run that last day. Also tasks depending on task a will have a that dependency satisfied for all days up until the latest run.

[martingrayson]

I've been playing with this by having a start date of September 1st and schedule_interval=timedelta(7). When running the scheduler for the first time I saw a backfill over this date range - i.e. multiple task runs, one for each day that has elapsed.

What I'd actually like is one task run (rather than a months worth), as its inefficient for my process to run multiple times. Is there not a way of either marking the dag as success or skipping an execution if the rundate isn't the maximum possible rundate? I guess the first scenario I describe is when the dag is seeded, but how about if my host has problems and my dag fails to run for a month? I'd like to see a single dag execution in this scenario.

[martingrayson]

Sorry for hijacking this thread but I thought I'd drop this here in case anyone else has the same requirement or I'm doing something insane (likely).

As my job is weekly and is a bulk process that cant be incrementally run using 7 dag executions, I've used the PythonBranchOperator to skip certain dag runs. Essentially I check {ds} when the dag is kicked off and branch to a DummyOperator if {ds} isnt yesterdays date. This means my weekly dag is has 6 dummy runs and 1 logical run when its scheduled.

[r39132]

It feels like this thread might have died.

We have a use-case for running daily DB snapshots using Airflow. We are running DB backups/snapshotting in Airflow, so that we can coordinate a DB snapshot with pausing DAGs that depend on that DB while it is down. If the DB snapshot DAG is paused for 3 days and the unpaused, we don't want it to run the DB snapshot for the past 3 days.. We have a work-around by checking the logic in the BashOperator itself, but I am wondering if we can leverage the only_run_latest idea as well.

[criccomini]

+1 This feature would be really useful. Basically, any DAG that doesn't use incremental date-based partitioning would benefit from this.

A trivial example is a DAG that simply snapshots a full DB and loads it into a BLOB store. If it runs once, you're caught up. If you have the DAG set to run hourly, and miss an hour for whatever reason, there's no point in snapshotting the DB and loading it back-to-back. Just run the latest one, and skip the rest.

[r39132]

To address a question asked by @thibault-ketterer, this issue differs from #262

Only_run_latest will be used for a job like a daily db snapshot job. Typically, this type of job needs to run once a day. If it doesn't run for 3 days because the DAG was paused, when the DAG is unpaused, the job should not run 3 times. It should only run for the latest time interval.

What I propose in #262 is to mark entries manually as "skipped or do not run" so that the scheduler or backfill job does not attempt them. This could be because we know there is a data problem in the time range.

[r39132]

I'm starting to look into an implementation. It seems that this functionality could be implemented by skipping DAG Runs that are not the latest. Unfortunately, the current handling of "Skipped" is such that it is not compatible with depends_on_past=True. So, if we skipped a bunch of DAG Runs and if the DAG specified depends_on_past=True, the latest would never be queued. It seems that we first need the system to support depends_on_past=True for skipped tasks.

This is a major roadmap item and is tracked in #1155

[pkexcellent]

Hi, I really need to use the only_run_latest feature, but I do not know how to use is in my code. Is this a feature has not beed developed or I just missed it? Can you give me a example of task declaration ? If this feature is not developed, how can I implement my code to realize the only_run_latest feature? Thanks!

[rproepp]

This feature would be really important for my use case as well. If I understand this correctly, Airflow currently cannot be used to replace a simple cronjob (without workarounds) because it will always try to backfill.

[dkromm]

Any update on this - my use-case needs an option like this in the operator

[liuhenry]

+1 to this feature! However, in the interim, you can kind of get a workaround using the BranchPythonOperator (@martingrayson mentioned something similar):

def current_date_function(ds, **kwargs):
    if date.today() == datetime.strptime(ds, '%Y-%m-%d').date():
        return 'do_something'
    else:
        return 'do_nothing'

check_current_date = BranchPythonOperator(
    task_id='check_current_date',
    provide_context=True,
    python_callable=current_date_function,
    dag=dag)

do_nothing = DummyOperator(
    task_id='do_nothing',
    dag=dag)

do_something = PostgresOperator(
    task_id='do_something',
    dag=dag)

check_current_date.set_downstream(do_nothing)
check_current_date.set_downstream(do_something)

[benjiao]

+1 on this feature. This will be very useful for snapshot/update-to-latest type tasks as @r39132 mentioned.