Create your first Airflow DAG (2024)

Let's start creating a Hello World workflow, which does nothing other than sending "Hello World!" to the log.

A DAG file, which is basically just a Python script, is a configuration file specifying the DAG’s structure as code.

The following are the steps by step to write an Airflow DAG or workflow:

1. Creating a python file
2. Importing the modules
3. Default Arguments for the DAG
4. Instantiate a DAG
5. Creating a callable function
6. Creating Tasks
7. Setting up Dependencies
8. Verifying the final Dag code
9. Test the Pipeline
10. Running the DAG

Step 1: Creating a python file

• Create the${AIRFLOW_HOME}/dagsdirectory if it is not present. Under${AIRFLOW_HOME}/dagsdirectory we can put the script in it (including python script, shell, sql, etc., to facilitate scheduling).

mkdir -p ${AIRFLOW_HOME}/dags 

• Create a new python filehello_world_dag.pyinside the${AIRFLOW_HOME}/dags/directory.

HelloWorld Application Structure:

${AIRFLOW_HOME}├── airflow.cfg├── airflow.db├── dags <- Your DAGs directory│ └── hello_world_dag.py <- Your DAG definition file└── unittests.cfg 

Add the following steps content to thehello_world_dag.pyfile.

vi ${AIRFLOW_HOME}/dags/hello_world_dag.py

Step 2: Importing the modules

• Import the"timedelta, datetime"module fromdatetimepackage to help us schedule the dags.
• Import the"DAG"module fromairflowpackage to instantiate the DAG object.
• Import theDummyOperatorfrom the"airflow.operators.dummy_operator"module.
• Import thePythonOperatorfrom the"airflow.operators.python_operator"module.

#datetimefrom datetime import timedelta, datetime# The DAG objectfrom airflow import DAG# Operatorsfrom airflow.operators.dummy_operator import DummyOperatorfrom airflow.operators.python_operator import PythonOperator 

Step 3: Default Arguments for the DAG

• Default arguments are passed to a DAG as default_args dictionary.
• This makes it easy to apply a common parameter to many operators without having to type it many times.

# initializing the default argumentsdefault_args = {'owner': 'Ranga','start_date': datetime(2022, 3, 4),'retries': 3,'retry_delay': timedelta(minutes=5)} 

Step 4: Instantiate a DAG

• Next we will instantiate a DAG object by passing the"dag_id"string which is the unique identifier of the dag.
• It is recommended to keep the python file name and"dag_id"same, so we will assign the"dag_id"as"hello_world_dag".

# Instantiate a DAG objecthello_world_dag = DAG('hello_world_dag',default_args=default_args,description='Hello World DAG',schedule_interval='* * * * *', catchup=False,tags=['example, helloworld']) 

Example usage:

- Daily schedule:- `schedule_interval='@daily'`- `schedule_interval='0 0 * * *'` 

• By default, Airflow will run tasks for all past intervals up to the current time. This behavior can be disabled by setting thecatchupparameter of a DAG to false, in which case Airflow will only start executing tasks from the current interval.

Step 5: Creating a callable function

• We also need to create a function that will be called by thePythonOperatoras shown below:

# python callable functiondef print_hello():return 'Hello World!' 

Step 6: Creating Tasks

• An objectinstantiatedfrom anoperatoris called atask. There are various types of operators available but we will first focus onDummyOperatorandPythonOperator.
• APythonOperatoris used tocall a python functioninside your DAG. We will create a PythonOperator object that calls a function which will return 'Hello World!' upon it's call.
• Like a DAG object has"dag_id", a PythonOperator object has a"task_id"which acts as it's identifier.
• It also has"python_callable"parameter which takes the name of the function to be called as it's input.

# Creating first taskstart_task = DummyOperator(task_id='start_task', dag=hello_world_dag)# Creating second taskhello_world_task = PythonOperator(task_id='hello_world_task', python_callable=print_hello, dag=hello_world_dag)# Creating third taskend_task = DummyOperator(task_id='end_task', dag=hello_world_dag) 

Step 7: Setting up Dependencies

• Set the dependencies or the order in which the tasks should be executed.
• We can set the dependencies of the task by writing thetask namesalong with>>or<<to indicate thedownstreamorupstreamflow respectively.
• Here are a few ways you can define dependencies between them:

# Set the order of execution of tasks. start_task >> hello_world_task >> end_task 

Step 8: Verifying the final DAG code

After compiling all the elements of the DAG, ourfinal codeshould look like this:

cat ${AIRFLOW_HOME}/dags/hello_world_dag.py

#datetimefrom datetime import timedelta, datetime# The DAG objectfrom airflow import DAG# Operatorsfrom airflow.operators.dummy_operator import DummyOperatorfrom airflow.operators.python_operator import PythonOperator# initializing the default argumentsdefault_args = {'owner': 'Ranga','start_date': datetime(2022, 3, 4),'retries': 3,'retry_delay': timedelta(minutes=5)}# Instantiate a DAG objecthello_world_dag = DAG('hello_world_dag',default_args=default_args,description='Hello World DAG',schedule_interval='* * * * *', catchup=False,tags=['example, helloworld'])# python callable functiondef print_hello():return 'Hello World!'# Creating first taskstart_task = DummyOperator(task_id='start_task', dag=hello_world_dag)# Creating second taskhello_world_task = PythonOperator(task_id='hello_world_task', python_callable=print_hello, dag=hello_world_dag)# Creating third taskend_task = DummyOperator(task_id='end_task', dag=hello_world_dag)# Set the order of execution of tasks. start_task >> hello_world_task >> end_task 

This file creates a simple DAG with just two operators, theDummyOperator, which does nothing, and aPythonOperatorwhich calls theprint_hellofunction when its task is executed.

Step 9: Test the Pipeline

Check the DAG file contains valid Python code by executing the file with Python:

python3 ${AIRFLOW_HOME}/dags/hello_world_dag.py 

Step 10: Running the DAG

• Activate the virtual environment.

source ~/install/airflow-tutorial/airflow_venv/bin/activateexport AIRFLOW_HOME=~/install/airflow-tutorial/airflow 

• Start the airflow webserver and scheduler.

airflow webserver \--pid ${AIRFLOW_HOME}/logs/airflow-webserver.pid \--stdout ${AIRFLOW_HOME}/logs/airflow-webserver.out \--stderr ${AIRFLOW_HOME}/logs/airflow-webserver.out \-l ${AIRFLOW_HOME}/logs/airflow-webserver.log \-p 8080-Dairflow scheduler \--pid ${AIRFLOW_HOME}/logs/airflow-scheduler.pid \--stdout ${AIRFLOW_HOME}/logs/airflow-scheduler.out \--stderr ${AIRFLOW_HOME}/logs/airflow-scheduler.out \-l ${AIRFLOW_HOME}/logs/airflow-scheduler.log \-D 

• The DAG should run successfully. In order to check thegraph viewortree view, you can hover over links and selectGraphorTreeoptions.

• You can also view thetask's execution informationusinglogs. To do so, simply,click on the task, and you should see the following dialog box:

• Next, click on theLogbutton, and you will be redirected to the task's log.

Happy Learning.