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

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

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

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

Hint

Use print() with the exact message.

Practice

1. What is the main purpose of manually triggering a DAG in Airflow?

easy

A. To stop a running DAG

B. To schedule a DAG to run automatically at a set time

C. To delete a DAG from Airflow

D. To start a workflow immediately with optional custom parameters

Solution

1. Step 1: Understand manual trigger concept

   Manual triggers allow users to start a workflow whenever they want, not waiting for a schedule.

2. Step 2: Recognize parameter use

   Manual triggers can include parameters to customize the run, making the workflow flexible.

3. Final Answer:

   To start a workflow immediately with optional custom parameters -> Option D

4. Quick Check:

   Manual trigger = immediate start + parameters [OK]

Hint: Manual trigger means start now, not later [OK]

Common Mistakes:

• Confusing manual trigger with scheduled runs
• Thinking manual trigger stops a DAG
• Assuming manual trigger deletes DAG

2. Which of the following is the correct way to access a parameter named 'user_id' passed during a manual trigger inside a DAG task?

easy

A. user_id = dag_run.conf.get('user_id')

B. user_id = dag_run.conf['user_id']

C. user_id = dag_run.params['user_id']

D. user_id = dag_run.get('user_id')

Solution

1. Step 1: Recall parameter access method

   Parameters passed during manual trigger are accessed via dag_run.conf dictionary.

2. Step 2: Use safe access method

   Using dag_run.conf.get('user_id') safely returns the value or None if missing, avoiding errors.

3. Final Answer:

   user_id = dag_run.conf.get('user_id') -> Option A

4. Quick Check:

   Use dag_run.conf.get() to read parameters [OK]

Hint: Use .conf.get() to safely get parameters [OK]

Common Mistakes:

• Using square brackets without .get() causing KeyError
• Using non-existent dag_run.params
• Calling dag_run.get() which is invalid

3. Given this DAG task code snippet inside a PythonOperator:

def greet(**context):
    name = context['dag_run'].conf.get('name', 'Guest')
    print(f"Hello, {name}!")

What will be printed if the DAG is manually triggered with parameters {'name': 'Alice'}?

medium

A. Hello, Guest!

B. Hello, Alice!

C. Hello, None!

D. Error: 'name' not found

Solution

1. Step 1: Understand parameter retrieval

   The code uses dag_run.conf.get('name', 'Guest') which returns 'Alice' if passed, else 'Guest'.

2. Step 2: Apply given parameters

   Since 'name' is provided as 'Alice', the print statement outputs "Hello, Alice!".

3. Final Answer:

   Hello, Alice! -> Option B

4. Quick Check:

   Parameter 'name' passed = 'Alice' printed [OK]

Hint: Default used only if parameter missing [OK]

Common Mistakes:

• Ignoring default value and expecting error
• Assuming parameter is missing
• Confusing print output with variable name

4. You wrote this code inside a DAG task:

def process(**kwargs):
    param = kwargs['dag_run'].conf['param']
    print(param)

When manually triggering the DAG without passing 'param', what happens?

medium

A. Raises KeyError exception

B. Prints empty string

C. Prints None

D. Prints 'param'

Solution

1. Step 1: Analyze parameter access method

   Accessing dag_run.conf['param'] without .get() causes KeyError if 'param' is missing.

2. Step 2: Understand missing parameter effect

   Since 'param' is not passed, the code raises KeyError and does not print anything.

3. Final Answer:

   Raises KeyError exception -> Option A

4. Quick Check:

   Missing key with [] causes KeyError [OK]

Hint: Use .get() to avoid KeyError on missing params [OK]

Common Mistakes:

• Expecting None or empty string instead of error
• Assuming print outputs 'param' string
• Not realizing [] access throws error

5. You want to manually trigger a DAG that processes orders. The DAG expects a parameter 'order_id'. You want to ensure the DAG uses 'unknown' if 'order_id' is missing. Which code snippet inside the DAG task correctly implements this?

hard

A. order_id = dag_run.conf['order_id'] or 'unknown'

B. order_id = dag_run.conf['order_id'] if dag_run.conf else 'unknown'

C. order_id = dag_run.conf.get('order_id', 'unknown')

D. order_id = dag_run.conf.get('order_id') or 'unknown'

Solution

1. Step 1: Understand default parameter handling

   Using dag_run.conf.get('order_id', 'unknown') returns 'order_id' if present, else 'unknown'.

2. Step 2: Compare options for correctness

   A fails with KeyError if missing. B checks if conf exists (empty {} is falsy), but if other parameters present conf is truthy causing KeyError. C works for missing but overrides falsy values like empty string. D is clean, safe, and handles all cases correctly.

3. Final Answer:

   order_id = dag_run.conf.get('order_id', 'unknown') -> Option C

4. Quick Check:

   Use .get(key, default) for safe defaults [OK]

Hint: Use .get(key, default) to set fallback values [OK]

Common Mistakes:

• Using [] without checking key existence
• Using or without default in .get()
• Overcomplicating with if-else instead of .get()