DAG Introduction

In Airflow, DAG stands for "Directed Acyclic Graph" and is one of the most important concepts in the platform. A DAG is used to define a workflow, which consists of the execution order and dependencies of a series of tasks, as well as their scheduling and monitoring.

In Airflow, each DAG is composed of a set of tasks (Task) that can be executed sequentially, in parallel, or in a hybrid manner. Tasks within a DAG are defined using Operators; each Operator represents a single task within the DAG and defines the execution logic for that task. For example, BashOperator represents executing a Shell command on a local or remote system, PythonOperator represents executing a Python function, and EmailOperator represents sending emails, among others.

A DAG can be defined using a Python script, where the DAG's name, description, schedule, and dependencies between tasks are specified. Airflow parses these Python scripts to generate DAG objects, which are then added to Airflow's metadata database. DAG objects can be visualized in the Airflow Web UI, and their scheduling and monitoring can be managed through the Web UI.

Tasks within a DAG can have their execution order determined by setting dependencies between them; for example, one task can be scheduled to run after another completes, or multiple tasks can be executed simultaneously. Airflow calculates the execution order of tasks based on these dependencies and submits the tasks to an Executor for execution.

Example

Here is a simple DAG example that includes three tasks:

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 6, 24),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'my_dag',
    default_args=default_args,
    description='A simple DAG',
    schedule_interval=timedelta(days=1),
)

task1 = BashOperator(
    task_id='task_1',
    bash_command='echo "Hello World from Task 1"',
    dag=dag,
)

task2 = BashOperator(
    task_id='task_2',
    bash_command='echo "Hello World from Task 2"',
    dag=dag,
)

task3 = BashOperator(
    task_id='task_3',
    bash_command='echo "Hello World from Task 3"',
    dag=dag,
)

task1 >> task2 >> task3

In this DAG, we define three tasks of type BashOperator, namely task1, task2, and task3. Their respective commands output "Hello World from Task 1", "Hello World from Task 2", and "Hello World from Task 3". The dependency relationship between these tasks is that task2 will only start execution after task1 completes, and task3 will only start execution after task2 completes. These dependencies are defined using the operator (>>).

Additionally, we define some metadata for the DAG, such as default arguments, the DAG's name and description, and its schedule. In this example, the DAG is scheduled to run once daily.

Schedule

In Airflow, a Schedule (scheduler) refers to the plan for when tasks should run. Airflow uses a structure called a DAG (Directed Acyclic Graph) to organize tasks and utilizes a Schedule to specify when each task should run.

The Schedule in Airflow consists of the following three elements:

  1. Start Date: The time when the first instance of the task should run, typically specified as a specific date.
  2. Interval: The frequency of task execution, which can be hourly (once per hour), daily (once per day), weekly (once per week), and so on.
  3. End Date: The time when the last instance of the task should run, typically also specified as a specific date.

In Airflow, you can use Python's datetime library to specify the start and end dates, and you can also use timedelta to specify the task's execution interval.

For example, the following code defines a task Schedule where the task runs once daily at 1:00 AM, starting from June 24, 2023, until it ends on June 30, 2023:

from datetime import datetime, timedelta
from airflow import DAG

dag = DAG(
    dag_id='my_dag',
    start_date=datetime(2023, 6, 24),
    end_date=datetime(2023, 6, 30),
    schedule_interval=timedelta(days=1),
)

This Schedule will run once daily at 1:00 AM until it ends on June 30, 2023.