Apache Airflow is likely one of the hottest orchestration instruments within the information subject, powering workflows for corporations worldwide. Nevertheless, anybody who has already labored with Airflow in a manufacturing setting, particularly in a posh one, is aware of that it may sometimes current some issues and bizarre bugs.
Among the many many points you should handle in an Airflow setting, one crucial metric usually flies beneath the radar: DAG parse time. Monitoring and optimizing parse time is crucial to keep away from efficiency bottlenecks and make sure the right functioning of your orchestrations, as we’ll discover on this article.
That stated, this tutorial goals to introduce airflow-parse-bench, an open-source instrument I developed to assist information engineers monitor and optimize their Airflow environments, offering insights to cut back code complexity and parse time.
Relating to Airflow, DAG parse time is commonly an missed metric. Parsing happens each time Airflow processes your Python recordsdata to construct the DAGs dynamically.
By default, all of your DAGs are parsed each 30 seconds — a frequency managed by the configuration variable min_file_process_interval. Because of this each 30 seconds, all of the Python code that’s current in your dags folder is learn, imported, and processed to generate DAG objects containing the duties to be scheduled. Efficiently processed recordsdata are then added to the DAG Bag.
Two key Airflow parts deal with this course of:
Collectively, each parts (generally known as the dag processor) are executed by the Airflow Scheduler, making certain that your DAG objects are up to date earlier than being triggered. Nevertheless, for scalability and safety causes, it’s also doable to run your dag processor as a separate element in your cluster.
In case your setting solely has a couple of dozen DAGs, it’s unlikely that the parsing course of will trigger any type of drawback. Nevertheless, it’s widespread to seek out manufacturing environments with lots of and even hundreds of DAGs. On this case, in case your parse time is just too excessive, it may result in:
- Delay DAG scheduling.
- Enhance useful resource utilization.
- Setting heartbeat points.
- Scheduler failures.
- Extreme CPU and reminiscence utilization, losing assets.
Now, think about having an setting with lots of of DAGs containing unnecessarily complicated parsing logic. Small inefficiencies can rapidly flip into vital issues, affecting the soundness and efficiency of your total Airflow setup.
When writing Airflow DAGs, there are some vital finest practices to remember to create optimized code. Though you could find quite a lot of tutorials on the right way to enhance your DAGs, I’ll summarize among the key rules that may considerably improve your DAG efficiency.
Restrict Prime-Degree Code
One of the vital widespread causes of excessive DAG parsing instances is inefficient or complicated top-level code. Prime-level code in an Airflow DAG file is executed each time the Scheduler parses the file. If this code contains resource-intensive operations, resembling database queries, API calls, or dynamic process era, it may considerably influence parsing efficiency.
The next code reveals an instance of a non-optimized DAG:
On this case, each time the file is parsed by the Scheduler, the top-level code is executed, making an API request and processing the DataFrame, which might considerably influence the parse time.
One other vital issue contributing to sluggish parsing is top-level imports. Each library imported on the prime degree is loaded into reminiscence throughout parsing, which could be time-consuming. To keep away from this, you’ll be able to transfer imports into capabilities or process definitions.
The next code reveals a greater model of the identical DAG:
Keep away from Xcoms and Variables in Prime-Degree 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 said 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 will 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, relatively than making a number of Variable.get() calls. Alternatively, use Jinja templates, as variables retrieved this manner are solely processed throughout process execution, not throughout DAG parsing.
Take away Pointless DAGs
Though it appears apparent, it’s at all times vital to recollect to periodically clear up pointless DAGs and recordsdata out of your setting:
- Take away unused DAGs: Examine your
dagsfolder and delete any recordsdata which might be not wanted. - Use
.airflowignore: Specify the recordsdata Airflow ought to deliberately ignore, skipping parsing. - Assessment paused DAGs: Paused DAGs are nonetheless parsed by the Scheduler, consuming assets. If they’re not required, contemplate eradicating or archiving them.
Change Airflow Configurations
Lastly, you may change some Airflow configurations to cut back the Scheduler useful resource utilization:
min_file_process_interval: This setting controls how usually (in seconds) Airflow parses your DAG recordsdata. Rising 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 thedagslisting for brand new DAGs. If you happen to deploy new DAGs occasionally, contemplate growing this interval to cut back CPU utilization.
We’ve mentioned loads concerning the significance of making optimized DAGs to keep up a wholesome Airflow setting. However how do you truly measure the parse time of your DAGs? Happily, there are a number of methods to do that, relying in your Airflow deployment or working system.
For instance, when you’ve got 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 simple, measuring the effectiveness of your code optimizations could be much less so. Each time you modify your code, you should redeploy the up to date Python file to your cloud supplier, look ahead to 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
Nevertheless, whereas easy, this strategy 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 instrument makes it simple to retailer parse instances, examine outcomes, and standardize comparisons throughout your DAGs.
Putting in the Library
Earlier than set up, it’s really useful to make use of a virtualenv to keep away from library conflicts. As soon as arrange, you’ll be able to set up the package deal by operating the next command:
pip set up airflow-parse-bench
Be aware: This command solely installs the important dependencies (associated to Airflow and Airflow suppliers). You should manually set up any extra libraries your DAGs rely upon.
For instance, if a DAG makes use of boto3 to work together with AWS, be certain that boto3 is put in in your setting. In any other case, you may 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 have to outline them domestically 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 Device
After putting in the library, you’ll be able to start measuring parse instances. 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 instances can range barely throughout runs on account of system and environmental components:
With a view 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 influence of the aforementioned optimizations, I changed the code in mydag_test.py with the optimized model shared earlier. After executing the identical command, I obtained the next end result:
As observed, 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 modifications we made!
There are different attention-grabbing options that I feel it’s related to share.
As a reminder, when you’ve got any doubts or issues utilizing the instrument, you’ll be able to entry the whole documentation on GitHub.
Apart from that, to view all of the parameters supported by the library, merely run:
airflow-parse-bench --help
Testing A number of DAGs
Usually, you seemingly 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 recordsdata 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
Working 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. Nevertheless, 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 could be particularly helpful throughout growth. Because the identify suggests, this feature 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 data, together with metrics and historical past, is saved in an area 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 have been the primary execution.
Parse time is a crucial metric for sustaining scalable and environment friendly Airflow environments, particularly as your orchestration necessities grow to be more and more complicated.
For that reason, the airflow-parse-bench library could be an vital instrument 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 quicker and extra performant.
Because the code is executed domestically, 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 setting.
Lastly, this undertaking is open for collaboration! When you’ve got strategies, concepts, or enhancements, be at liberty to contribute on GitHub.
