The Ultimate 101 Guide to Apache Airflow DAGS (2024)

Looking for an efficient tool for streamlining and automating your data processing workflows? Apache Airflow DAGs are your one-stop solution! Read this blog till the end to learn everything you need to know about Airflow DAG.

Let's consider an example of a data processing pipeline that involves ingesting data from various sources, cleaning it, and then performing analysis. The workflow can be broken down into individual tasks such as data ingestion, data cleaning, data transformation, and data analysis. These tasks may have dependencies on each other, such as data cleaning depending on the successful completion of data ingestion. By defining a DAG, the dependencies can be clearly established and the workflow can be executed in the correct order, ensuring the accuracy and integrity of the final analysis. DAGs provide a way to represent the dependencies between the different tasks in a workflow. For example, if a task depends on the output of another task, the DAG can be used to define that relationship. This allows for complex workflows to be broken down into smaller, manageable tasks that can be executed in a specific order.

Managing complex data pipelines can be challenging, requiring coordination between multiple systems and teams. This is where Apache Airflow DAGs come in. DAG stands for Directed Acyclic Graphs. Apache Airflow DAGs provide a powerful tool for creating and managing data pipelines, streamlining the process of data processing and automation. With its open-source framework and modular design, Apache Airflow DAGs offer flexibility and scalability for data scientists, engineers, and analysts. Whether you need to extract, transform, or load data, Airflow DAGs provide a simple yet powerful way to manage your data pipeline. Airflow's flexibility and extensibility make it a popular choice for managing data pipelines across various industries, from finance and healthcare to media and entertainment. This blog will dive into the details of Apache Airflow DAGs, exploring how they work and multiple examples of using Airflow DAGs for data processing and automation workflows.

What are Apache Airflow DAGs?

Apache Airflow DAGs (Directed Acyclic Graphs) is a popular open-source tool for creating, scheduling, and monitoring data pipelines. Airflow DAGs represent a set of tasks that must be executed to complete a workflow. Each task is defined as a node in a graph, and the edges between nodes represent the dependencies between tasks.

Example of DAG in Airflow

One of the most common Apache Airflow example DAGs can be ETL (Extract, Transform, Load) pipelines, where data is extracted from one source, transformed into a different format, and loaded into a target destination.

Another Airflow DAG example could be automating the workflow of a data science team, where tasks such as data cleaning, model training, and model deployment can be represented as individual nodes in an Airflow DAG. Airflow DAGs provide a powerful and flexible way to manage complex workflows, making it easier to monitor and troubleshoot data pipelines and enabling organizations to process large amounts of data with ease.

Source: Airflow DAG Documentation

What is a DAG file in Airflow?

A DAG file in Airflow stands for Directed Acyclic Graph file. It is a Python script that defines and organizes tasks in a workflow. The DAG file specifies the order in which tasks should be executed and their dependencies, allowing for efficient scheduling and monitoring of data pipelines in Airflow.

Core Concepts of Airflow DAGs

The following are the core concepts in Airflow DAGs:

• Task: is a basic unit of work in an Airflow Directed Acyclic Graph. A series of tasks organized together, based on their dependencies, forms Airflow DAG. It is represented as a node in DAG and is written in Python.

• Operator: They are building blocks of Airflow DAGs. Airflow operators hold the data processing logic. Each task in a DAG is an instance of an operator. There are many operators available in Airflow. E.g., the Python operator executes Python code, and the Snowflake operator executes a query against the Snowflake database.

• Sensors: They are a subclass of Operators that waits for an event to occur. They only wait for something to happen (a file to enter or time-based) and then pass the execution to the downstream task.

• Hooks: They are abstractions to low-level API connecting to external systems. They help connect to external systems like HDFS, S3, PostgreSQL, etc., easily without writing much boilerplate code. Airflow also provides an interface to create custom hooks without built-in hooks.

Airflow DAGs Architecture

The following are key components in Apache Airflow DAGs:

• Scheduler: It is a Python process responsible for scheduling jobs. It determines what tasks need to run, when they need to run, and where to run.

• Executor: They manage the running of tasks allocated by the Scheduler. There are two types of Executors:
  1. Local executors run on the same server as the Scheduler.
  2. Remote executors operate on different machines, allowing them to scale out and distribute jobs across computers(Workers).

• Web Server: A Flask server that serves the Airflow UI (user interface). It helps in monitoring, triggering, and debugging DAGs and their tasks.

• Metadata Database: It stores past and current DAG runs, DAG configurations, and other metadata information. By default, it is an SQLite database, but you can choose from PostgreSQL, MySQL, and MS SQL databases.

• DAG directory: It is a folder of DAG files. It is read by the scheduler and executor.

Source: Airflow Documentation

How To Create Airflow DAGs?

Once Airflow is installed, and the database is initiated, the following steps will help create a simple Apache Airflow DAG. In this example, a single DAG uses multiple tasks with Python and Bash Operators to display current DateTime, with dependency established between them.

Step 1: Create a Python file
Under {$AIRFLOW_HOME}/dags folder, create a Python file named my_first_dag.py. $AIRFLOW_HOME represents the path where Airflow has been installed. The dag code must be present in the dags folder. Airflow loads dag objects from Python files stored in the above directory.

Step 2: Import required modules
The following modules are required to create the entire DAG:

• DAG module to instantiate DAG object.

• Python operator to create Python task.

• Bash operator to run Bash command.

• Datetime module to display current DateTime.

Step 3: Create the Airflow DAG instance
A DAG instance must have two parameters:

• dag id: An identifier of a dag that must be unique across all of the dags.

• start_date: The logical date at which the dag starts to get scheduled. It is also referred to as the execution date.

schedule interval and catchup are other dag parameters. Schedule interval defines when the DAG needs to get triggered while catch=False prevents backfilling. It prevents the creation of DAG runs from strat_date to the current date.
There are multiple ways to declare a DAG: using Context Manager (as shown in the below image), using the standard constructor, or using @dag decorator to turn a Python function into a DAG generator.

Step 4: Create Tasks
Here, we are going to define two task instances. Every task must have task ids defined and they must be unique among other tasks.

• Python Task: The Python task runs a Python function. Therefore, we create a Python function that returns the current date time. This function is specified in the python_callable argument.

• Bash Task: The Bash task executes Bash commands. Here, it just says that the pipeline ran successfully.

Build a Job Winning Data Engineer Portfolio with Solved End-to-End Big Data Projects.

Step 5: Defining DAG dependencies
The last step is connecting the two DAG tasks (upstream tasks and downstream tasks), i.e., establishing dependency between them. Here, upstream tasks and downstream tasks are defined, thus specifying the order in which the different tasks must run. There are two ways of defining DAG dependencies:

• By using bitwise operators, i.e., >> or <<

• Using set_upstream and set_downstream methods.

On running the above-created DAG in the UI or when DAG runs at a scheduled interval, we get the below output on successful execution. The below image also shows the DAG structure with two tasks. The triggered DAG and all the tasks can be viewed in Airflow UI. Airflow Logs show task instance details for debugging purposes.

Apache Airflow Dynamic DAGs

This section will give you an overview of Airflow Dynamic DAGs and how to create them.

Why Use Apache Airflow Dynamic DAGs?

Sometimes, writing a static DAG is not practical. Consider a use case where you need to load data into tables from external APIs. Each table is loaded from a different API. If there are hundreds of tables, writing the same DAG code for each table with few parameter changes is not ideal. It is time-consuming. It is harder to maintain as the same change must be repeated in each DAG code. This problem is solved by Dynamic DAGs.

Creating Apache Airflow Dynamic DAGs

In Airflow, DAG references are stored in globals(). In Python, globals() refers to a built-in function that returns a dictionary containing global variables. Once the DAG object is stored in globals(), Airflow loads the DAG. There are two methods to generate DAGs dynamically:

• Single File Method: It is the simplest way to generate Dynamic DAG. In this method, inside a single file, you write a function that returns a DAG object and a for loop to call the function multiple times. Each time the function is called, required DAG parameters are passed, and the DAG object is stored in globals(). This method is recommended if there are few simple DAG instances to generate.

• Multiple File Method: In this method, a template file is created, defining the DAG with placeholders. Each DAG has a JSON file (input file) defining the values of the placeholder. A Python script reads each of the input files, replaces the placeholders in the template file, and generates the DAG instance. This method is reliable for generating multiple complex DAG instances.

What Are The Different Ways to Visualize DAGs in Apache Airflow?

Apache Airflow provides several ways to visualize DAGs (Directed Acyclic Graphs) to make it easier for developers and data engineers to understand and manage their data pipelines. One of the most simple ways to visualize a DAG is using the Airflow web UI, which provides a graphical representation of the DAG, showing the dependencies between tasks and their status. Another way to visualize DAGs in Airflow is by using the command-line interface, which allows developers to see the status of individual tasks and their DAG dependencies in real-time. Airflow also provides the option to export DAGs to a JSON file format, which can be used to visualize DAGs using external tools such as Graphviz. Additionally, Airflow supports integration with third-party visualization tools such as Gantt charts and BI (Business Intelligence) tools like Tableau and Power BI.

Airflow DAG Python

Airflow DAGs can be defined using Python, allowing developers to take advantage of the powerful capabilities of Python for data processing and analysis. With Airflow DAG Python, developers can define the tasks and dependencies of a data pipeline using simple Python code, which makes it easy to automate complex workflows. Python-based Airflow DAGs can be customized to meet the specific needs of a data processing task, with access to a rich ecosystem of Python libraries and tools for data analysis, such as pandas and NumPy. Additionally, Python-based Airflow DAGs can be easily integrated with other Python-based applications and services, creating end-to-end data processing pipelines that seamlessly move data between applications.

Apache Airflow DAGs Folder

In Apache Airflow, a DAGs folder is a directory where Airflow searches for Python files containing DAG definitions. The DAGs folder is specified in the Airflow configuration file, and by default, it is located in the ~/airflow/dags directory. Any Python file containing a DAG definition in the DAGs folder will be automatically detected by Airflow and added to the list of available DAGs in the Airflow web UI. The DAGs folder allows developers to organize their DAG definitions in a consistent and structured way, making it easier to manage and maintain their data pipelines.

Apache Airflow DAG Dependencies

Apache Airflow DAG Dependencies refer to the relationships between tasks in a Directed Acyclic Graph (DAG). These Airflow task dependencies determine the order in which tasks are executed and ensure that each task is completed successfully before the next one begins.

There are two types of dependencies in Apache Airflow:

• Upstream Dependencies- tasks that must complete successfully before the current task can start.

• Downstream Dependencies- tasks that can only start after the current task has been completed successfully.

Airflow provides several operators to define dependencies between tasks in a DAG, including the BashOperator, PythonOperator, and more.

In addition to these operators, Airflow also provides a set of sensors that allow tasks to wait for specific conditions to be met before they start. For example, a FileSensor waits for a file to appear in a specific location before starting the next task in the DAG.

In Apache Airflow, you can also define dependencies between DAGs using the ExternalTaskSensor. This allows you to create complex workflows across multiple DAGs.

Apache Airflow DAG Arguments

In Apache Airflow, DAG (Directed Acyclic Graph) arguments are used to define and configure the DAG tasks. DAG arguments can be passed to the constructor of the DAG class and include DAG parameters such as DAG ID, default arguments, start date, schedule interval, and concurrency. The DAG ID is a unique identifier for the DAG and is used to reference the DAG throughout Airflow. The default arguments specify the default parameters that will be used by all the tasks in the DAG unless explicitly overridden. The start date determines the time when the DAG will start executing tasks. The schedule interval defines how often the DAG will run- hourly, daily, or weekly. The concurrency parameter determines the maximum number of tasks running concurrently in the DAG.

Other DAG arguments include the catchup parameter, which determines if the DAG will run tasks for previous dates if it starts later than the start date, and the orientation parameter, which determines the direction of the DAG from the start date. Additionally, DAG arguments can set up email notifications, configure retry policies for failed tasks, and define dependencies between tasks.

How to Test Airflow DAGs?

Airflow DAGs must be tested extensively before deploying to production. Since DAGs are written in Python, any of its test runners, like unittest, and pytest can be used for testing Airflow DAGs. The following tests can be carried out to ensure DAGs are production ready:

• DAG initialisation test
  These tests ensure DAG objects have the correct structure i.e. they are acyclic and without import DAG errors. Such tests can be carried out in a local Airflow environment. Run the Python script containing the DAG definition to test whether DAG objects are getting created without syntax errors. To test for import DAG errors, use DagBag() function from the Airflow models package. The DagBag object has an import_errors parameter which holds the total number of import DAG errors.

• Unit Testing
  In software development, unit testing is a method to test small chunks of source code individually to ensure they return desired output. In Airflow, unit tests are usually written to test the logic and functionality of Hooks and Operators. It is important for unit tests to pass before deploying code to production.

• Integration tests and end-to-end pipeline tests are also written to test the correct functioning of the entire DAG run.

Tips for Troubleshooting and Debugging Airflow DAG

Once the DAG is said to run, we come across many issues. Following are the common errors encountered while running Airflow DAGs and their possible solutions:

• DAG does not run on schedule
  This is a common issue while running DAGs. The first step in resolving this issue is to check whether the scheduler is running. You will see an error in Airflow UI if the scheduler is not running. An easy fix is to run the command airflow scheduler in the command line. Another fix would be to check if at least one schedule interval period is passed. The start date and schedule interval must be passed as an argument.

Source: DataCamp

• DAGs won't load
  You will often see that the new DAG won't appear in the web UI. First, check if the Python file containing the DAG definition is in the correct DAG folder. You can check the path of DAG's folder in airflow.cfg file. Another reason may be the presence of Syntax errors. Running the airflow list_dags command will output debugging information, which helps in troubleshooting further.

• Tasks are not running
  Though DAG is visible in UI yet tasks are not running when the DAG is triggered. One solution is to check if DAG is in paused state. If DAG is paused, tasks don't run. If your tasks are scheduled or queued, ensure the scheduler is running correctly. If required, restart the scheduler or increase its resources.

Whenever a DAG runs, logs are generated, and they are visible in the Airflow UI. These log messages are the best resource for debugging and troubleshooting errors in Airflow data pipelines.

Best Practices for Designing and Organizing DAG in Airflow

DAGs must be idempotent. An idempotent task returns the same output for a given input on running it multiple times. DAGs can be made idempotent by designing each task in DAG to be idempotent. This property helps to recover quickly in case of failure of DAGs.
The following design principles help in building idempotent, efficient, and scalable DAGs:

• DAG Tasks Must be Atomic
  Each task in the DAG must perform one operation independently. For Example: In an ETL pipeline, it is recommended to create three separate tasks, one each for Extract, Transform, and Load operations. Upon failure of a task, it can be re-run independently.

• Top-Level Code Must be Avoided
  In Airflow, top-level code refers to any code that is not part of DAG creation. E.g.: Code that makes requests to the external database. Airflow, by default, scans the DAG folder every 30 seconds to update with new changes. The presence of top-level code would try to connect to an external system every 30 seconds instead of just once when the DAG is set to run, affecting the performance.

• Maintain Minimum Code
  Consider a DAG file to be similar to a config file. Having code that isn't part of DAG creation is difficult and harder to maintain. If your DAG uses SQL script or Python function, place them in a separate file. Use the template_searchpath parameter while defining DAG to include the external files.

• Make use of Provider Packages
  Provider packages are additional packages created by Airflow Community. It is recommended to use these packages rather than writing your Python function to connect to third-party data processing tools, as these are readily available and tend to be more efficient.

• Using a Single Method for Defining DAG Dependencies
  DAG dependencies can be defined in two ways:
  1. Using bitwise operator >> and <<
  2. Using set_upstream() and set_downstream() functions
  It is a personal choice to choose either method, but it is important to pick one method for consistency and readability.

Worried about finding good Hadoop projects with Source Code? ProjectPro has solved end-to-end Hadoop projects to help you kickstart your Big Data career.

Airflow DAGs Examples and Project Ideas

The following is a list of projects involving Apache Airflow. These projects will help in understanding the different use cases where Airflow could be used for data processing.

1. Consuming Weather API And Storing in PostgreSQL Database

In this Airflow project, Apache Airflow is used to extract the last 5 days of data from the Open Weather API of 10 different locations. The extracted data is sent to the Spark cluster for processing and generating aggregated values. The generated values are stored in Postgre SQL, and materialized views are created to view the results. The entire data pipeline is orchestrated using Airflow. GitHub repository link

2. Football Fantasy Premier League (FPL) Project

Fantasy League has become a very popular online sport, and people are using data to make informed decisions in building their teams. This project involves building a data pipeline that pulls FPL data from an API and loads it into the Postgre SQL database, where the data is analyzed to pick a football team for Fantasy League. The entire pipeline is orchestrated using Airflow, and the DAG runs weekly to pull in new data. GitHub repository Link

3. Music Streaming Project

This is another end-to-end project using Apache Airflow. A stream of generated events is processed in real-time and ingested into cloud storage Data Lake. Apache Airflow manages data processing tasks, applying transformations on data present in Data Lake and creating required tables for publishing a dashboard to generate analytics. GitHub repository Link

4. AWS Snowflake Data Pipeline Example using Kinesis and Airflow

Working on this project will teach you how to create a Snowflake Data Pipeline, starting from the EC2 logs to storage in Snowflake and S3 post-transformation and processing through Airflow DAGs. In this project, two streams of data- customers data and orders data will be added to Snowflake and S3 processed stage through Airflow DAG processing and transformation.

Source link- AWS Snowflake Data Pipeline Example using Kinesis and Airflow

Automate Your Data Pipelines Using Apache Airflow DAGs With ProjectPro

Apache Airflow DAGs have become an essential tool for data engineers and developers who need to automate data pipelines and manage complex workflows. By using Airflow, they can create DAGs that define tasks and their dependencies, schedule when they should run, and monitor their progress. Working on real-world Airflow projects can help developers gain hands-on experience in creating and managing DAGs, implementing best practices for data privacy, monitoring and troubleshooting pipelines, and integrating Airflow with external systems. At ProjectPro, we offer a variety of real-world projects that leverage Apache Airflow DAGs, allowing developers to learn and work with this powerful tool. By mastering Airflow, developers can improve their skills and expertise and help their organizations achieve their data management and analysis goals.

FAQs on Apache Airflow DAGs

1. How many DAGs can Airflow have?

There is no specific limit on the number of DAGs defined in Apache Airflow. The number of DAGs that can be created depends on the available system resources and the complexity of the DAGs.

2. What are DAGs used for?

Apache Airflow DAGs are commonly used for data processing and workflow automation tasks, including ETL (Extract, Transform, Load) processes, machine learning pipelines, and data analysis workflows. They can also schedule and orchestrate any repeatable tasks, including database backups, API calls, and web scraping.

3. What are sensors in Apache Airflow?

Sensors in Apache Airflow are operators that wait for a certain condition to be satisfied before allowing a DAG to proceed. They are useful for triggering workflows based on external events or resources.

4. What are operators in Apache Airflow?

Operators in Apache Airflow are building blocks that represent a single task in a workflow. Each operator performs a specific action, such as executing SQL queries, transferring data between systems, or triggering external events, and can be combined to create complex workflows.

5. How do you schedule a DAG in Apache Airflow?

You can schedule a DAG in Apache Airflow by defining the schedule_interval parameter in the DAG definition. This parameter specifies the frequency with which the DAG should be executed, such as daily or hourly, and can be set using cron notation or a timedelta object.

6. How do you pass parameters to your operators in Apache Airflow?

You can pass parameters to operators in Apache Airflow using the 'params' argument in the DAG file or the 'provide_context' argument to access task instance details.

7. How do you use a sensor in your DAG in Apache Airflow?

You can use a sensor in your DAG in Apache Airflow by defining it as a task and specifying its parameters, such as the poke interval and timeout. The sensor task will wait until the specified condition is met before proceeding to the next task in the DAG.