Manual triggers and parameters in Apache Airflow - Mini Project: Build & Apply

Manual Triggers and Parameters in Airflow

📖 Scenario: You are managing data workflows using Apache Airflow. Sometimes you want to run a workflow manually and pass specific information to control its behavior.For example, you want to run a data processing task manually and tell it which data source to use.

🎯 Goal: Build a simple Airflow DAG that can be triggered manually with a parameter called source. The DAG will print the value of this parameter.

📋 What You'll Learn

Create a DAG with the id manual_trigger_dag

Add a source parameter to the DAG's run configuration

Use a PythonOperator to print the source parameter value

Ensure the DAG can be triggered manually with the parameter

💡 Why This Matters

🌍 Real World

In real projects, manual triggers with parameters let you run workflows on demand with custom inputs, like processing specific files or dates.

💼 Career

Understanding manual triggers and parameters is essential for DevOps engineers and data engineers managing workflows in Airflow.

1

Create the initial DAG structure

Create a DAG called manual_trigger_dag with a daily schedule and default arguments including start_date set to January 1, 2024.

Apache Airflow

# Your code here

Need a hint?

Use DAG from airflow and set dag_id, default_args, and schedule_interval.

2

Add a Python function to print the parameter

Define a Python function called print_source that accepts **kwargs and prints the value of source from kwargs['dag_run'].conf.

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

default_args = {
    'start_date': datetime(2024, 1, 1)
}

dag = DAG(
    dag_id='manual_trigger_dag',
    default_args=default_args,
    schedule_interval='@daily'
)

# Define print_source function here
# Your code here

Need a hint?

Use kwargs['dag_run'].conf.get('source') to get the parameter.

3

Add a PythonOperator task using the function

Create a PythonOperator task called print_source_task that uses the print_source function and add it to the DAG.

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

default_args = {
    'start_date': datetime(2024, 1, 1)
}

dag = DAG(
    dag_id='manual_trigger_dag',
    default_args=default_args,
    schedule_interval='@daily'
)

def print_source(**kwargs):
    source = kwargs['dag_run'].conf.get('source')
    print(f"Source parameter value: {source}")

# Create PythonOperator task print_source_task here
# Your code here

Need a hint?

Use PythonOperator with task_id, python_callable, and dag arguments.

4

Print the task id to confirm setup

Write a print statement to display the string 'DAG manual_trigger_dag is ready with task print_source_task'.

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

default_args = {
    'start_date': datetime(2024, 1, 1)
}

dag = DAG(
    dag_id='manual_trigger_dag',
    default_args=default_args,
    schedule_interval='@daily'
)

def print_source(**kwargs):
    source = kwargs['dag_run'].conf.get('source')
    print(f"Source parameter value: {source}")

print_source_task = PythonOperator(
    task_id='print_source_task',
    python_callable=print_source,
    dag=dag
)

# Print confirmation message below
# Your code here

Need a hint?

Use print() with the exact message.