Stop Creating Bad DAGs — Optimize Your Airflow Environment By Improving Your Python Code | by Alvaro Leandro Cavalcante Carneiro | Jan, 2025

Keep away from Xcoms and Variables in High-Stage Code

Nonetheless speaking about the identical matter, is especially attention-grabbing to keep away from utilizing Xcoms and Variables in your top-level code. As acknowledged by Google documentation:

If you’re utilizing Variable.get() in high stage code, each time the .py file is parsed, Airflow executes a Variable.get() which opens a session to the DB. This may dramatically decelerate parse instances.

To deal with this, think about using a JSON dictionary to retrieve a number of variables in a single database question, reasonably than making a number of Variable.get() calls. Alternatively, use Jinja templates, as variables retrieved this fashion are solely processed throughout job execution, not throughout DAG parsing.

Take away Pointless DAGs

Though it appears apparent, it’s all the time vital to recollect to periodically clear up pointless DAGs and information out of your surroundings:

• Take away unused DAGs: Test your dags folder and delete any information which are now not wanted.
• Use .airflowignore: Specify the information Airflow ought to deliberately ignore, skipping parsing.
• Overview paused DAGs: Paused DAGs are nonetheless parsed by the Scheduler, consuming sources. If they’re now not required, take into account eradicating or archiving them.

Change Airflow Configurations

Lastly, you might change some Airflow configurations to scale back the Scheduler useful resource utilization:

• min_file_process_interval: This setting controls how usually (in seconds) Airflow parses your DAG information. Growing it from the default 30 seconds can scale back the Scheduler’s load at the price of slower DAG updates.
• dag_dir_list_interval: This determines how usually (in seconds) Airflow scans the dags listing for brand spanking new DAGs. In the event you deploy new DAGs sometimes, take into account growing this interval to scale back CPU utilization.

We’ve mentioned lots in regards to the significance of making optimized DAGs to keep up a wholesome Airflow surroundings. However how do you truly measure the parse time of your DAGs? Thankfully, there are a number of methods to do that, relying in your Airflow deployment or working system.

For instance, you probably have a Cloud Composer deployment, you’ll be able to simply retrieve a DAG parse report by executing the next command on Google CLI:

gcloud composer environments run $ENVIRONMENT_NAME 
— location $LOCATION 
dags report

Whereas retrieving parse metrics is easy, measuring the effectiveness of your code optimizations will be much less so. Each time you modify your code, it’s essential to redeploy the up to date Python file to your cloud supplier, anticipate the DAG to be parsed, after which extract a brand new report — a sluggish and time-consuming course of.

One other doable strategy, in case you’re on Linux or Mac, is to run this command to measure the parse time domestically in your machine:

time python airflow/example_dags/instance.py

Nonetheless, whereas easy, this strategy just isn’t sensible for systematically measuring and evaluating the parse instances of a number of DAGs.

To deal with these challenges, I created the airflow-parse-bench, a Python library that simplifies measuring and evaluating the parse instances of your DAGs utilizing Airflow’s native parse technique.

The airflow-parse-bench software makes it simple to retailer parse instances, evaluate outcomes, and standardize comparisons throughout your DAGs.

Putting in the Library

Earlier than set up, it’s really helpful to make use of a virtualenv to keep away from library conflicts. As soon as arrange, you’ll be able to set up the bundle by operating the next command:

pip set up airflow-parse-bench

Observe: This command solely installs the important dependencies (associated to Airflow and Airflow suppliers). You will need to manually set up any extra libraries your DAGs rely on.

For instance, if a DAG makes use of boto3 to work together with AWS, be certain that boto3 is put in in your surroundings. In any other case, you will encounter parse errors.

After that, it’s a necessity to initialize your Airflow database. This may be performed by executing the next command:

airflow db init

As well as, in case your DAGs use Airflow Variables, you need to outline them domestically as nicely. Nonetheless, it’s not vital to place actual values in your variables, because the precise values aren’t required for parsing functions:

airflow variables set MY_VARIABLE 'ANY TEST VALUE'

With out this, you’ll encounter an error like:

error: 'Variable MY_VARIABLE doesn't exist'

Utilizing the Instrument

After putting in the library, you’ll be able to start measuring parse instances. For instance, suppose you’ve gotten a DAG file named dag_test.py containing the non-optimized DAG code used within the instance above.

To measure its parse time, merely run:

airflow-parse-bench --path dag_test.py

This execution produces the next output:

As noticed, our DAG offered a parse time of 0.61 seconds. If I run the command once more, I’ll see some small variations, as parse instances can differ barely throughout runs because of system and environmental elements:

In an effort to current a extra concise quantity, it’s doable to mixture a number of executions by specifying the variety of iterations:

airflow-parse-bench --path dag_test.py --num-iterations 5

Though it takes a bit longer to complete, this calculates the common parse time throughout 5 executions.

Now, to guage the impression of the aforementioned optimizations, I changed the code in mydag_test.py with the optimized model shared earlier. After executing the identical command, I acquired the next end result:

As observed, simply making use of some good practices was able to lowering virtually 0.5 seconds within the DAG parse time, highlighting the significance of the adjustments we made!

There are different attention-grabbing options that I feel it’s related to share.

As a reminder, you probably have any doubts or issues utilizing the software, you’ll be able to entry the whole documentation on GitHub.

In addition to that, to view all of the parameters supported by the library, merely run:

airflow-parse-bench --help

Testing A number of DAGs

Typically, you probably have dozens of DAGs to check the parse instances. To deal with this use case, I created a folder named dags and put 4 Python information inside it.

To measure the parse instances for all of the DAGs in a folder, it is simply essential to specify the folder path within the --path parameter:

airflow-parse-bench --path my_path/dags

Operating this command produces a desk summarizing the parse instances for all of the DAGs within the folder:

By default, the desk is sorted from the quickest to the slowest DAG. Nonetheless, you’ll be able to reverse the order through the use of the --order parameter:

airflow-parse-bench --path my_path/dags --order desc

Skipping Unchanged DAGs

The --skip-unchanged parameter will be particularly helpful throughout improvement. Because the identify suggests, this selection skips the parse execution for DAGs that have not been modified for the reason that final execution:

airflow-parse-bench --path my_path/dags --skip-unchanged

As proven under, when the DAGs stay unchanged, the output displays no distinction in parse instances:

Resetting the Database

All DAG info, together with metrics and historical past, is saved in a neighborhood SQLite database. If you wish to clear all saved information and begin contemporary, use the --reset-db flag:

airflow-parse-bench --path my_path/dags --reset-db

This command resets the database and processes the DAGs as if it had been the primary execution.

Parse time is a crucial metric for sustaining scalable and environment friendly Airflow environments, particularly as your orchestration necessities change into more and more complicated.

Because of this, the airflow-parse-bench library will be an vital software for serving to information engineers create higher DAGs. By testing your DAGs’ parse time domestically, you’ll be able to simply and rapidly discover your code bottleneck, making your dags sooner and extra performant.

Because the code is executed domestically, the produced parse time gained’t be the identical because the one current in your Airflow cluster. Nonetheless, if you’ll be able to scale back the parse time in your native machine, the identical could be reproduced in your cloud surroundings.

Lastly, this challenge is open for collaboration! In case you have solutions, concepts, or enhancements, be at liberty to contribute on GitHub.