Dependencies on past dates of tasks with a SequentialExecutor

[thoralf-gutierrez]

Hey guys !

I'm trying to fit my tasks in the airflow world, but I'm hitting some walls...

I have normal "daily" tasks that should run daily, and then I have "range" tasks that should run only when I have the result of the last 30 days of those daily tasks. There can only be one of those daily + range task running at once: I'm using a SequentialExecutor.

Right now, I'm especially interested in doing this in a backfill mode. I'd like to say : run these DAGs for 6 months of data, and I want it to understand that :

• it can run all those daily tasks in any order (depends_on_past is false)
• it can run any range task whenever all of its dependencies are satisfied (depends_on_past is also false here)

To some extent, the dependencies that I want to express for my range task is a mix of what I can do with a normal set_downstream and an extended depends_on_past where I could set a number of dates rather than being limited to "the previous date".

I've tried dividing the daily tasks and the range tasks in two DAGs, with sensors in the range DAG that wait for the (external) daily tasks to succeed (one sensor per date on which the range task depends). This works until one of the Sensors that doesn't have all his dependencies satisfied blocks the SequentialExecutor.

I've tried putting everything in the same DAG, using priorities to tell him what to run first, but priorities don't have an effect on backfill jobs.

Am I missing another possible solution ?

If that's not the case, what would be the extension that would best fit airflow:

• Implementing priorities in the BackfillJob class ?
• Extending set_downstream with relative dates (similarly to the ExternalTaskSensor or the depends_on_past) ?

Happy to contribute with any of these if they make sense ;-)

Cheers,
Thoralf

[mistercrunch]

The SequentialExecutor probably shouldn't be used beyond tutorial and early playing around. I wrote it for testing purposes and to allow early bootstrapping while using sqlite. Setup a mysql/postgres and graduate to the local executor!

[thoralf-gutierrez]

I'm not sure I was perfectly explicit: I can unfortunately only run one of the daily or range tasks at a time. This is why we were using a SequentialExecutor at first.

Your remark got me thinking though. Actually, I only want the daily and range tasks to run one at a time, but I'd like to have any number of external task sensors running/waiting at the same time. This sounds like I could use a LocalExecutor with one pool for the daily and range tasks with max parallelism of one, and one pool for the external task sensors with unlimited parallelism.

Unfortunately, I see that the BackfillJob class doesn't look at those pools either => Would it make sense to add something similar to the SchedulerJob.prioritize_queued method to the BackfillJob class or was it omitted on purpose ?

[mlimotte]

I have a similar use-case. I do it like this:

• Use LocalExecutor (to avoid a 'wait' task from blocking everything)
• separate into 2 DAGs (daily vs. the range)
• For the daily tasks, set depends_on_past: True. Even if it's not entirely true
• Have my "range" task wait on the last period of the daily task using an ExternalTaskSensor

depends_on_past: True may not be literally true for your job, but it provides two behaviors as a consequence. 1) The daily jobs execute one at a time, 2) when the final task of your "range" executes you know that all previous tasks are done.

[mistercrunch]

Yes backfill doesn't play nice with pools currently. It'll get fixed soon.

[thoralf-gutierrez]

Thanks @mlimotte - I hadn't thought of that and I think it will indeed do the trick !

@mistercrunch : Probably, what I really want is to have an extended set_downstream with relative dates (similarly to the ExternalTaskSensor or the depends_on_past). I want to be able to say "This task depends on the run of this task for this date". The dependency would be explicit (which is not the case with sensors). Would this feature make sense ? If it's the case, I'd be happy to implement the backend.

[mistercrunch]

ExternalTaskSensor allows to implement "This task depends on the run of this task for this date" where "this date" is relative to the current date :
http://pythonhosted.org/airflow/code.html#airflow.operators.ExternalTaskSensor

[thoralf-gutierrez]

What I'm trying to say is that ExternalTaskSensors don't create explicit dependencies : they are not directly accounted for in TaskInstance.are_dependencies_met. Hence I need to introduce ad-hoc solutions to my problem, either by spawning many ExternalTaskSensor that are continuously waiting or by introducing fake depends_on_past as proposed by @mlimotte.

I'm imagining an extended set_downstream that would allow me to express "This task depends on the run of this task (same DAG) for this date" as an explicit dependency rather than through adhoc solutions. Would this extended set_downstream make sense in airflow or is it something definitely out of scope because it doesn't fit airflow's paradigm?

Thanks for your quick replies!

[sv3ndk]

The functionality described by Thoralf above boils down to the ability to define a DAG with operators depending on several previous executions of operators within the same DAG. For example, a daily trend batch (1) and a trend diff batch (2) that processes the results of (1) over the last 7 days. In that example, (2) should be scheduled for the first time only after 7 days of results of (1) are available.

At the moment, an operator can only declare an upstream dependency towards another operator with identical execution date (via set_upstream() ), or towards itself with execution date being the latest previously scheduled (via depends_on_past=True).

Airflow also allows to build the operator (2) with a start_date explicitly set to 7 days after the start_that of (1), but this seems to be a concern leaking from the Airflow scheduler to the developer designing the DAG: ideally Airflow would resolve that start_date by itself. Also, this start_date does not enforce that the execution of (2) will happen after all the required executions of (1) are finished.

It seems that ExternalTaskSensor (with several DAGs) cannot be used for that purpose either because it has no direct influence on the scheduling but rather allows the corresponding task(s) to be launched and kept waiting. In the example above, that would yield 7 instances of task (2) in case of a LocalExecutor or CeleriExecutor, or even just one of them at a time in case of a SequencialExecutor until one of them blocks forever, which would halt the whole task sequence.

Our suggested resolution would be a PR along he following lines:

• in DAG, add an execution_calendar object which relies on utils.date_range and self.schedule_interval to provide the execution dates to be scheduled (among others, required in BackfillJob) as well as the past execution dates. + adapt BackfillJob and others accordingly.
• in DAG and BaseOperator, adapt the set_dependency, set_downstream and others to allow to declare a number of past upstream executions (defaulting to 1), possibly with new DownstreamDependency and UpstreamDependency classes. We would not declare if those past executions need to be successful or not, since this concern is already captured by task.trigger_rule.
• in TaskInstance.are_dependencies_met(), adapt the SQLAlchemy query to rely on the DAG execution_calendar as well as on the number of past upstream executions of each upstream dependency to query the status of depended upon tasks.

[mistercrunch]

A few notes:

• start_date is necessary for the scheduler to know where to start. Also for change management as you introduce new tasks to a DAG, it's an important piece of metadata that the scheduler can't guess
• start_date also informs depends_on_past to not be applied on that first date

It does make sense to add more functionality to the dependency engine, and time ranges is reasonable. There's already a fairly complex flow for understanding "why isn't my task getting triggered" and I want to make sure that we keep or increase the level of clarity around that as we make the engine more flexible.

I think "Trigger Rules" might become the place to store rules more complex than "this task depends on this task". In its most complex form, you could have arbitrarily complex rules that could be combined in complex ways. Picture something like trigger_rules=(DependOnPast(self) and FullDagExecution('some_other_dag')) or AllHellBreaksLoose().

In the meantime it's pretty easy to write custom sensors, what about an alternative to ExternalTaskSensor that can take date ranges as input and be start_date aware?

[thoralf-gutierrez]

Thanks for your answer !

Indeed, we see where you coming from for the start_date, thanks for clarifying.

We're happy to see that you're also interested in more functionalities such as the time ranges.

We totally agree with the need for clarity. We were actually thinking about adding some justification to the task state that could be displayed in the UI. Maybe those justifications would contain reasons for success, failure or still queued, etc.

On the main topic : If we understand correctly, you'd like to gather the concepts of upstream / downstream dependencies, depends_on_past and trigger rules (aka all_success, any_failure, etc) into a flexible grammar. This grammar would allow to parametrize upstreams dependencies with time ranges and expected status (success, failure, etc) and to compose them with logical operators.

Here's a link to an overview of a suggested refactoring of the API which includes all those concerns as well as the task runnable state and justification that could be displayed in the UI. We're aware of the importance of API backwards compatibility, more thoughts will probably have to be thrown in there to address it.

https://gist.github.com/thoralf-gutierrez/00555b2a667099cfcfcb

To add some more context we are developing data analytics solutions on Spark and we are looking for something very similar to Airflow. We would definitely prefer to contribute the necessary effort to make it more flexible instead of developing a new in-house solution!

If you're available, we'd be happy to have a chat with you to discuss the topic and how we could collaborate on it! (see my email on my github profile)

[mistercrunch]

Nice. The proposal captures key elements.

We need a base class that can be extended to define new triggers. Evaluate would receive the context (as operators execute method does). Maybe overloading or/and would be nice but that can be tricky.

Then you can extend this base class and create anything from AllSuccessTriggerRule (receives a list of task_ids) to DependOnPastTriggerRule, to LocalWeatherIsGoodEnoughTriggerRule. Generally I prefer a wide collection of TriggerRules that are more targeted (TaskDateRangeTriggerRule) over a few with a complex list of parameters.

We'll have to preserve the previous API somehow.

I've also been thinking about a feature in the UI that would break down why a task isn't getting triggered, having this more modular framework would make it easy to show the list of rules and their current status.

[thoralf-gutierrez]

Closing the issue on GH, I've summarized and posted this on the mailing list https://groups.google.com/forum/#!topic/airbnb_airflow/SAu6gKi_GOo