Cease Creating Dangerous DAGs — Optimize Your Airflow Surroundings By Enhancing 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 prime degree code, each time the .py file is parsed, Airflow executes a Variable.get() which opens a session to the DB. This could dramatically decelerate parse occasions.

To handle this, think about using a JSON dictionary to retrieve a number of variables in a single database question, quite than making a number of Variable.get() calls. Alternatively, use Jinja templates, as variables retrieved this fashion are solely processed throughout activity 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 recordsdata out of your atmosphere:

• Take away unused DAGs: Examine your dags folder and delete any recordsdata which might be now not wanted.
• Use .airflowignore: Specify the recordsdata 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, think about eradicating or archiving them.

Change Airflow Configurations

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

• min_file_process_interval: This setting controls how typically (in seconds) Airflow parses your DAG recordsdata. 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 typically (in seconds) Airflow scans the dags listing for brand new DAGs. For those who deploy new DAGs occasionally, think about growing this interval to scale back CPU utilization.

We’ve mentioned quite a bit concerning the significance of making optimized DAGs to keep up a wholesome Airflow atmosphere. However how do you really measure the parse time of your DAGs? Luckily, there are a number of methods to do that, relying in your Airflow deployment or working system.

For instance, in case you have a Cloud Composer deployment, you possibly can 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, you must redeploy the up to date Python file to your cloud supplier, look forward to the DAG to be parsed, after which extract a brand new report — a gradual and time-consuming course of.

One other attainable strategy, if you happen to’re on Linux or Mac, is to run this command to measure the parse time regionally in your machine:

time python airflow/example_dags/instance.py

Nevertheless, whereas easy, this strategy shouldn’t be sensible for systematically measuring and evaluating the parse occasions of a number of DAGs.

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

The airflow-parse-bench instrument makes it straightforward to retailer parse occasions, evaluate outcomes, and standardize comparisons throughout your DAGs.

Putting in the Library

Earlier than set up, it’s advisable to make use of a virtualenv to keep away from library conflicts. As soon as arrange, you possibly can set up the package deal by operating the next command:

pip set up airflow-parse-bench

Word: This command solely installs the important dependencies (associated to Airflow and Airflow suppliers). You have 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 atmosphere. In any other case, you may encounter parse errors.

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

airflow db init

As well as, in case your DAGs use Airflow Variables, you have to outline them regionally as effectively. Nevertheless, it’s not mandatory 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 Software

After putting in the library, you possibly can start measuring parse occasions. For instance, suppose you could have 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 occasions can fluctuate barely throughout runs resulting from system and environmental elements:

With a purpose to current a extra concise quantity, it’s attainable 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 outcome:

As seen, simply making use of some good practices was able to decreasing 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 believe it’s related to share.

As a reminder, in case you have any doubts or issues utilizing the instrument, you possibly can entry the entire 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

Normally, you seemingly have dozens of DAGs to check the parse occasions. To handle this use case, I created a folder named dags and put 4 Python recordsdata inside it.

To measure the parse occasions 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 occasions for all of the DAGs within the folder:

By default, the desk is sorted from the quickest to the slowest DAG. Nevertheless, you possibly can 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 feature skips the parse execution for DAGs that have not been modified because the final execution:

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

As proven beneath, when the DAGs stay unchanged, the output displays no distinction in parse occasions:

Resetting the Database

All DAG data, together with metrics and historical past, is saved in a neighborhood SQLite database. If you wish to clear all saved information and begin recent, 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 develop into more and more complicated.

For that reason, the airflow-parse-bench library will be an vital instrument for serving to information engineers create higher DAGs. By testing your DAGs’ parse time regionally, you possibly can simply and rapidly discover your code bottleneck, making your dags quicker and extra performant.

For the reason that code is executed regionally, the produced parse time received’t be the identical because the one current in your Airflow cluster. Nevertheless, if you’ll be able to scale back the parse time in your native machine, the identical could be reproduced in your cloud atmosphere.

Lastly, this mission is open for collaboration! When you’ve got solutions, concepts, or enhancements, be at liberty to contribute on GitHub.