how to efficiently make airflow dag definitions database-driven

Question

Background

I have some dags that pull data from an 3rd-party api.

The accounts we need to pull can change over time. To determine which accounts to pull, depending on the process we may need to query a database or make an HTTP request.

Before airflow, we would just get the account list at the start of the python script. Then we would iterate through the account list and pull each account to file or whatever it was we needed to do.

But now, using airflow, it makes sense to define tasks at the account level and let airflow handle retry functionality and date range and parallel execution etc.

Thus my dag might look something like this:

Problem

Since each account is a task, the account list needs to be accessed with every dag parse. But since dag files are parsed frequently, you don't necessarily want to query the database or wait for a REST call with every dag parse from every machine all day long. This could be resource intensive, and could cost money.

Question

Is there a good way to cache this type of config information in a local file, ideally with a specified time-to-live?

Thoughts

I have thought about a couple different approaches:

1. write to csv or pickle file and use mtime to expire.
   • the concern with this is that i might get collisions if two processes try to expire the file at the same time. i don't know how likely this is or what the consequences would be but probably nothing terrible.
2. create a common sqlite DB for all such processes. should be auto created first time a variable is accessed. each config variable gets a row in table. use last_modified_datetime column to tell when to expire.
   • requires more elaborate code & dependencies.
3. use airflow variables
   • nice thing about this would be that it uses existing DB, so would be no $ per query and reasonable network lag, but it still requires network round trip.
   • has benefit of being identical across all nodes in a multi-node setup.
   • determining when to expire would probably be problematic so would probably create config manager dag to update the config variables periodically.
   • but then this would add complexity to deployment and devolpment process -- the variables need to be populated in order to define the DAGs properly -- all developers would need to manage this locally too, as opposed to a more create-on-read cacheing approach.
4. Subdags?
   • never used them, but I have a suspicion they could be used here. But the community seems to discourage their use anyway...

Have you dealt with this problem? Did you arrive at a good solution? None of these seems very good.

Answer 1

I have the same exact scenario.

Have API call for multiple accounts. Initially created a python script to iterate the list.

When I started using Airflow thought about what you are planning to do. Tried 2 of the alternatives you listed. After some experimentation decided to handle retry logic within python with simple try-except blocks if HTTP calls fail. Reasons are

1. One script to maintain
2. Less Airflow objects
3. Restartability is easier with one script in place. (restarting failed job in Airflow is not a breeze (no pun intended))

At the end it's up to you, that was my experience.

Answer 2

Airflow default DAG parsing interval is pretty forgiving: 5 minutes. But even that is quite a lot for most people, so it's quite reasonable to increase that if your deployment isn't too close to the due times for the new DAGs.

In general, I'd say it's not that bad to make a REST request at every DAG parse heartbeat. Also, nowadays the scheduling process is decoupled from the parsing process, so that won't affect how fast your tasks are scheduled. Airflow caches the DAG definition for you.

If you think you still have reasons to put your own cache on top of that, my suggestion is to cache at the definitions server, not on the Airflow side. For example, using cache headers on the REST endpoint and handling cache invalidation yourself when you need it. But that could be some premature optimization, so my advice is to start without it and implement it only if you measure convincing evidence that you need it.

EDIT: regarding Webserver and Worker

It's true that the Webserver will trigger DAG Parses as well, not sure about how frequent. Probably following the guicorn workers refresh interval (which is 30 seconds by default). Workers will do it also by default at the start of every task, but that can be saved if you activate pickling DAGs. Not sure if that's a good idea though, I've heard this is something destined to be deprecated.

One other thing you can try to do is to cache that in the Airflow process itself, memoizing the function that makes the expensive request. Python has a built-in functools for that (lru_cache) and together with pickling it might be enough and very very much easier than the other options.