Airflow Operators & Dependencies: Multi-Task DAGs

Setting Up the Environment

Open VS Code, go to the Airflow directory, then the DAG folder, and create a file named pipeline.py.

Imports and Functions

Import three operators: PythonOperator, BashOperator and EmptyOperator to set start and end markers. Also import the DAG class and datetime for the start date:

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
from datetime import datetime

Create three functions for validating, transforming, and reporting data:

def validate():
    print("Validating incoming data.")

def transform():
    print("Transforming data.")

def report():
    print("Creating report.")

DAG Definition

with DAG(
    dag_id="pipeline",
    start_date=datetime(2023, 1, 1),
    schedule="@daily",
    catchup=False
) as dag:

Creating Tasks

    start = EmptyOperator(task_id="start")

    validate_task = PythonOperator(
        task_id="validate",
        python_callable=validate
    )

    check_logs = BashOperator(
        task_id="check_logs",
        bash_command="echo 'Checking system logs.'"
    )

    transform_task = PythonOperator(
        task_id="transform",
        python_callable=transform
    )

    report_task = PythonOperator(
        task_id="report",
        python_callable=report
    )

    end = EmptyOperator(task_id="end")

Setting Dependencies

    start >> [validate_task, check_logs] >> transform_task >> report_task >> end

Save the file.

Running the DAG in the UI

Open the Airflow UI, enter credentials (username and password both airflow), and sign in. Go to the DAG section, search for pipeline, and open it. Click Trigger, choose Single Run.

Go to the Runs section to see the cycle was successful.

Checking Logs

Start and end are EmptyOperators, so they only print Airflow lifecycle messages. Click on one to see it inherits from EmptyOperator and has no logs.

For the validate task, the output shows "Validating incoming data." This line appears after the task and DAG names are returned.

The check_logs task is a BashOperator with slightly different lines because it runs a shell command. The output says "Checking system logs."