Dag grouping

[MatthewStrickland]

Description

Support grouping dags, where a dag can be part of many groups, with a configuration similar to max_active_runs across the whole group.

Use case/motivation

(I'm sure this must have been discussed somewhere before but I can't find any searches for the topic, apologies if there are)

In a nutshell: Motivations relating to the reasoning behind the current existence of DAG level max_active_runs.

Use case: My dags each have a 1-1 relationship to a single table in my schema which they aim to mutate. Occasionally I need to duplicate a dag with a static start/end date to fill in some missing data (eg. backfilling jobs for earlier dates, or backfilling jobs for existing dates without overriding existing data). All dags by themselves run with max_active_runs=1.
Each dag I have creates a snapshot of the table (and rolls back if necessary when things go wrong). If a rollback of this snapshot were to trigger it is pretty much guaranteed to have negative impacts on the persistence of the data of another dag working on the same table.

What would you like to happen? Ideally the ability to make a DAGGroup object with max_active_runs. (I'm not sure if it's better for the DagGroup to add dags, or if the dag should specify which DagGroup(s) to belong to).

Alternatives:

• I can pause DAGs manually that might interfere with each other, but would need to actively monitor them
• Cross-DAG dependencies, though I feel all dags would need knowledge of each other and would be a nightmare to maintain and get right

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[mariotaddeucci]

Just for curiosity, did you try to create a pool for this scenario?
I have an issue based when i consume an API and hit the rate limit, pools are good to manage this.
Add the pool to default argument and you will have the same effect that you described.
https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/pools.html

[MatthewStrickland]

Just for curiosity, did you try to create a pool for this scenario? I have an issue based when i consume an API and hit the rate limit, pools are good to manage this. Add the pool to default argument and you will have the same effect that you described. https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/pools.html

@mariotaddeucci I hadn't looked at pools before but just took a look now. If I specify the pool (with 1 slot) at the DAG level (inside default_args I guess) will it block all other dags using the same pool from starting their execution until the first dag completes? That would suit me if so, except it would then force all my internal tasks to be non-async, which isn't ideal.

However if it applies only at the task level would dags potentially compete for the slot? In which case it might not be suitable because my ETL workload runs across several tasks.

[RNHTTR]

Could you use a cluster DAG policy? You can add the desired DAGs into a list object, and do something like...

def dag_policy(dag: DAG):
    """Configure max_active_runs for group of DAGs."""
    if dag.dag_id in ["dag_a", "dag_b"] :
        dag.max_active_runs = 1

[MatthewStrickland]

Could you use a cluster DAG policy? You can add the desired DAGs into a list object, and do something like...

def dag_policy(dag: DAG):
    """Configure max_active_runs for group of DAGs."""
    if dag.dag_id in ["dag_a", "dag_b"] :
        dag.max_active_runs = 1

It's the first time I've noticed this feature so thanks for pointing it out for me. My reading of it though is that it's just a way to define common config for all dags?
My ask is to have a way that dag_a cannot run at the same time as dag_b, unless I'm misreading this feature and this can accomplish that.

[potiuk]

Airlfow does not have any "per dag" enforcement. It works pretty much exclusively on "task" level. Scheduling is not based on DAGs it's based on task. Schedulers look at tasks and picks those tasks that are eligible to run. There are thing like depends_on_pst - https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html#depends-on-past that will make different instances of same task depend (and hold from being executed) on other instances, but there is no feature to have such a dependency on the whole DagRun level.

You can emulate it with Pools as mentioned above, and you can write your DAGs in a way to accomodate it. Assumption is there that usually those are individual tasks that shoud not run in parallel or with too many instances, not the whole dag runs.

Individual DAGs can have in-DAG limitations (such as max_active_tasks or you can define max_active_runs for it. The last case would do what you want for an indivdual DAG (i.e. Scheduler will not schedule a new DAG Run if there are are other DAG Runs of the same DAG running)

If you look hard enough and think about your concrete case, (but it very much depends on nature of your limitations and why you want to exclude running in parallel and what "real" problems you want to avoid by allowing some taks from different DAGs to run in parallel) - maybe with a combination of pools/dependencies, structure of the DAGs to achieve what you want. I think in vast majority of the cases this is possible.

What you are asking for @MatthewStrickland sounds like legitimate, albeit a bit niche case if you want to simplify your job of designing the DAGs / DAGRuns and think about them more like "whole entities" that can inteeract with each other - which is not the primary focus of Airlfow, because Airflow is Task based.

But - if you think it would be a valuable contribution - and something that could be usable for others as well, I think it would be fine to accept a contribution of such feature, if you you, your compan would like to contribute it - making your problem simpler to solve but also helping others in a similar situation. You are absolutely free to attempt it. Happy to review such PRs, it does not seem to be changing a lot Airflow's architecture so I do not think it needs bigger discussion or AIP, it just needs someone to roll their sleeves up and contribute it (or if those who want it do not feel strong enough to do it, even paying someone to contribute it).

That would be a nice way to give back to the community and become one of the 2800 or so contributors to Airflow - many features like that were added to Airflow because the users felt the need for it and became contributors.

So I heartily welcom you @MatthewStrickland and your company especially (since they are those who can likely decide on spending company resources on it) to pursue that, if the current model is not enough for you.

[potiuk]

Just as a good example and showing how it can be done - here is a great article (published today) of on of our users who contributed something they needed. A Journey from the User to Contributor https://medium.com/apache-airflow/bridging-gaps-in-airflow-unveiling-the-journey-from-user-to-contributor-54d361c75191 - you might want to read it @MatthewStrickland and see if that's a viable path for you.

[github-actions]

This issue has been automatically marked as stale because it has been open for 14 days with no response from the author. It will be closed in next 7 days if no further activity occurs from the issue author.

[github-actions]

This issue has been closed because it has not received response from the issue author.