Large number of tasks results in scheduler consuming excessive amounts of memory

[Traubert]

I have a situation with a large number (thousands) of tasks getting parsed and serialized ahead of time, and eventually my scheduler dies with OOM, strangely in json decoding, which takes around a gigabyte of ram within 30 minutes of startup. Flamegraph from memray here: https://a3s.fi/hardwick-2006633-flip/memray-flamegraph-memray.html.

Is there a way to postpone parsing with all these tasks ahead of time?

[potiuk]

What is the verion airflow do you use? Could you also elaborate a bit more - what kind of deployment you have, what is configurations, providers, what kind of DAGS you have, what are the sizes of dags stored in Serialized DAG tables?

Scheduler generally parses the dags continuosly and uses serizlied version of the DAG (SerializedDag) to perform all kind of operations - for example it uses it to perorm scheduling. As - such it will periodically load the SerializedDag in memory and if you have. huge DAG with thousands of tasks then it will have to be loaded in memory fully in order to make scheduling decisions. I wonder how big your DAG files are ? And can you see this growth to continue forever. It would be great to see a few flamegraphs - from different times when scheduler runs - to see if there is a constant growth coming from a single place.

[Traubert]

Hey, sorry I missed this, thanks a ton for getting back! We're running Airflow 2.7.1. Our config is in Ansible templates, see https://github.com/CSCfi/kielipankki-nlf-harvester/blob/main/ansible/roles/airflow-config/templates/airflow.cfg.j2. We're using LocalExecutor, everything is running in one VM. Our problematic dags are these downloads of a large number of files that are combined into packages: https://github.com/CSCfi/kielipankki-nlf-harvester/blob/main/pipeline/dags/download_images.py. We make download batches of a configurable size, and each download batch becomes a task, and thus we get thousands of tasks. When I log verbosely, it looks like Airflow is using a lot of resources checking over and over is these tasks could be started.

It's hard to get a longer view of memory use than I provided, because it just grows and grows until the scheduler is killed. It basically all comes from dagbag.py, here https://github.com/apache/airflow/blob/b73366799d98195a5ccc49a2008932186c4763b5/airflow/models/dagbag.py#L190C30-L190C30, where the DagBag method get_dag() does this:

            if dag_id not in self.dags:
                # Load from DB if not (yet) in the bag
                self._add_dag_from_db(dag_id=dag_id, session=session)
                return self.dags.get(dag_id)

Maybe it would ultimately finish doing this and have loaded everything, but I don't want to use gobs of memory just for this :).

We've ended up hobbling along with reducing the task count a bit and accepting periodic Airflow restarts.