Versioning Problem of (likely badly written) Dynamically Created DAGS

[gbonazzoli]

Apache Airflow version

3.0.4

If "Other Airflow 2 version" selected, which one?

No response

What happened?

I have a Python Source that generate PythonOperator Tasks on a DAG.

It is the load of the Staging Area in a Datawarehouse, so during this phase there are hundreds of tables that we need to load from the transactional database to the analytical one, Airflow limits the load through max_active_tasks=32 DAG parameter.

The code is:

    with TaskGroup(group_id="MOVEX100") as DLTHUB_MOVEX100:
        for source_table_list in [SOURCE_MOVEX100_TABLES]:
            for source_table_obj in source_table_list:
                table_name = source_table_obj["table_name"]
                dlt_query = source_table_obj["dlt_query"]

                dlt_load_movex100 = PythonOperator(
                    task_id="dlt_movex100_" + table_name,
                    python_callable=execute_dlt_movex100,
                    retries=3,  # Retry up to 3 times (original + 3 retries)
                    retry_delay=timedelta(minutes=randrange(5) + 1),  # Wait from 1 to 5 minutes between retries
                    templates_dict={
                        "table_name": f"{table_name}",
                        "dlt_query": f"{dlt_query}",
                    },
                )

                Beg_Load_Staging >> dlt_load_movex100 >> End_Load_Staging

The problem with this code is that every days Airflow 3.x registers hundreds of different versions for the DAG.

The reality is that is dynamically generated and so it is obvious that changes dynamically.

Is there anyway to limit the problem ?

What you think should happen instead?

No response

How to reproduce

I think that is a general problem afflicting Airflow 3.x with DAG that are dinamically generated through code.

Operating System

Ubuntu 24.04.3 LTS

Versions of Apache Airflow Providers

apache-airflow-providers-celery==3.12.2
apache-airflow-providers-cncf-kubernetes==10.6.2
apache-airflow-providers-common-compat==1.7.3
apache-airflow-providers-common-io==1.6.2
apache-airflow-providers-common-sql==1.27.4
apache-airflow-providers-fab==2.3.1
apache-airflow-providers-git==0.0.5
apache-airflow-providers-postgres==6.2.2
apache-airflow-providers-redis==4.2.0
apache-airflow-providers-salesforce==5.11.2
apache-airflow-providers-smtp==2.1.2
apache-airflow-providers-ssh==4.1.2
apache-airflow-providers-standard==1.5.0

Deployment

Virtualenv installation

Deployment details

An ubuntu vanilla installation in python 3.12 virtual env.

Anything else?

no

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

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

[tirkarthi]

Does using dynamic task mapping help here?

https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/dynamic-task-mapping.html

cc: @jedcunningham @ephraimbuddy

[potiuk]

The reality is that is dynamically generated and so it is obvious that changes dynamically.

That's not true in general.

If your intention to generate a different dag every time it is parsed, then yes it will be dynamically changing and. as far as I know - you can disable versioning https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#disable-bundle-versioning for that bundle if your bundle contains DAG that change frequently and it is expected (but this is generally a bad practice)

Dynamically generated dags are NOT supposed to dynamically change every time they are parsed. They MAY change slower or faster - depending on how often the "input" to those dynamic dags change. For example if you have list of countries, and you generate dynamic dags based on that - your parsed dag should change only when you add a new country - every time (providing that your code to generate dynamic dag is well written) , parsing the same Dag should produce the same - dynamically generated - parsed representation.

So in your case you likely:

a) your intention is to change your dag every time it is parssed (which is quite strange, but then you should disable versioning for that bundle)

b) there is a problem with dynamic dag generation code - it should produce the same parsed representation, every time Dag is parsed.

There might be two resons for b)

1. your dynamic dag generation has problems and does not produce stable dag object structure or stable parameters. This might change for example when you do not use fixed start date - but use days_ago or similar - which is a bad practice, because days_ago() will change every time your dag is parsed. Or you can pass time() or other dynamically generated (and ever changing) content as part of your dag definition. Or you create objects in a random sequence every time you build your dynamic dag. This might happen for example when you run "SELECT * FROM" and do not ORDER the select - because the databases return data in such select in random order in general, and the order might change any time.

2. there is a bug in our Dag serialization code

Most likely it's 1) and you have to fix your dynamic dag generation to be "stable" if you want to version dag. Which means that serialized version of your Dag should be THE SAME if Dag parsing on your Dag is executed - if none of the inputs changed - and they should remain stable over time, so they should be independent on the time when they are parsed.

It could also be 2) - but in order to know whether it's 1) or 2) you have to look at the serialized dag table (see https://airflow.apache.org/docs/apache-airflow/stable/database-erd-ref.html ) find two versions of your Dags and find out what's the difference betwen the dags and what kind of "stability" issue in dynamic dag generation causes it. The "compressed" representation in this table is compressed with .gzip so you need to retrieve the blobs and decompress them -> you will get .json representation of the dags that you should be able to compare and see where the problem is coming from.

Also it might be easier for you if you review your dynamic dag generation looking at potential stability issues from those described above.

Once you find the reasons - please let us know - it would be interesting to find out what stability issues people like you might have - or maybe you will find some of the problems coming from our serializaiton. So we are looking forward for your investigation.

Converting it into a discussion until you investigate it and provide more information.

[potiuk]

I also updated the title to properly reflect it - this is not a "huge problem with dag versioning", it's either lack of understanding that "dynamically generated dags" should have "stable" property if you want to version them or some specific bug in serializaiton code.

I hope as result of that discussion maybe you @gbonazzoli -> when you find reasons for that stability issues you can help us to write good documentation describing what "stability" property of such dynamically generated Dags is - or maybe we will even find some bugs in the implementation.

But this is NOT huge problenm with versioning in general, it works as intended, and it should work well with dynamically generated Dags - providing of course that those Dags are properly generated and that there are no bugs.

Looking forward to results of your investigation @gbonazzoli

[potiuk]

Maybe even - eventually - we might come up with better tooling to analyse such dynamic dag "stabilty" issues - but I find this case an interesting case study to see what could be the reasons, so that we can write our tooling to be useful to quickly analyse such issues.

[JoseBueno-plytix]

Thanks for your fast response.

When you say "any change", do you mean change in the parameters (like adding/removing parameters, which makes perfect sense) or do you refere to the value of said parameters? For example, a simplification of the DAG im currently testing for this, it recives a int as a parameter that then passes to a task to perform some action with that number, even when running the DAG with the same input, it creates a new version.

What i mean by direction is the memory direction where the DagParam object is located (DagParam object at 0x7fb7bf966750)

[potiuk]

Any change in the serialized form of the dag. In case you pass an object that is not serializable as parameter that is supposed to the string - it is converted to string which contains address to the object.

Don't do it.

[potiuk]

It does not do you any good when you pass an object to parameters that is supposed to be string. As simple as that.

[potiuk]

you can - for example - use mypy or other type checkers to verify if your code passes the right types of params

[JoseBueno-plytix]

Using pydantic for assuring the right types, maybe i am missing something, but i don't know why it is supossed to be an string, I define it a as an int