Manual triggers and parameters in Apache Airflow - Step-by-Step Execution

Process Flow - Manual triggers and parameters

User opens Airflow UI

↓

Select DAG to trigger

↓

Click 'Trigger DAG'

↓

Input parameters (optional)

↓

DAG run starts with parameters

↓

Tasks execute using parameters

↓

DAG run completes

The user manually triggers a DAG from the Airflow UI, optionally inputs parameters, which the DAG run then uses during execution.

Execution Sample

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago

def print_param(**kwargs):
    param = (kwargs['dag_run'].conf or {}).get('my_param', 'default')
    print(f"Parameter received: {param}")

dag = DAG('manual_trigger_example', start_date=days_ago(1))

run_task = PythonOperator(
    task_id='print_param_task',
    python_callable=print_param,
    dag=dag
)

This DAG prints a parameter passed during manual trigger or 'default' if none is given.

Process Table

Step	Action	Parameter Input	Parameter Value Used	Output
1	User opens Airflow UI and selects DAG	N/A	N/A	Ready to trigger
2	User clicks 'Trigger DAG' button	N/A	N/A	Trigger dialog opens
3	User inputs parameter 'my_param'='hello'	my_param=hello	hello	DAG run starts
4	Task 'print_param_task' executes	my_param=hello	hello	Parameter received: hello
5	DAG run completes	N/A	N/A	Success
6	User triggers DAG again without parameter	None	default	DAG run starts
7	Task 'print_param_task' executes	None	default	Parameter received: default
8	DAG run completes	N/A	N/A	Success

💡 Execution stops after DAG run completes successfully.

Status Tracker

Variable	Start	After Step 3	After Step 6	Final
my_param	N/A	hello	default	default or hello depending on trigger

Key Moments - 2 Insights

Why does the task print 'default' sometimes instead of the parameter I entered?

If no parameter is provided during manual trigger (see step 6 in execution_table), the code uses the default value 'default' as shown in step 7.

How does Airflow pass parameters to tasks during manual trigger?

Parameters are passed via the DAG run's conf dictionary, accessed in the task with kwargs['dag_run'].conf, as shown in step 4.

Visual Quiz - 3 Questions

Look at the execution_table at step 4. What parameter value does the task use?

A hello

B default

C None

D my_param

💡 Hint

Check the 'Parameter Value Used' column at step 4 in execution_table.

At which step does the DAG run start without any parameter provided?

A Step 3

B Step 6

C Step 1

D Step 8

💡 Hint

Look for 'None' in the 'Parameter Input' column in execution_table.

If the user inputs parameter 'my_param=world' during trigger, what will the task print?

A Parameter received: hello

B Parameter received: default

C Parameter received: world

D Parameter received: None

💡 Hint

Refer to how parameter values are used in execution_table steps 3 and 4.

Concept Snapshot

Manual triggers in Airflow allow users to start DAG runs from the UI.
Users can input parameters as JSON in the trigger dialog.
Tasks access parameters via dag_run.conf dictionary.
If no parameter is given, tasks can use default values.
This enables flexible, on-demand DAG execution.

Full Transcript

In Airflow, you can manually start a DAG run from the web interface. When you do this, a dialog lets you enter parameters as JSON. These parameters are passed to the DAG run and can be accessed by tasks through the dag_run.conf dictionary. If you don't provide parameters, tasks can use default values coded in the DAG. This lets you run workflows on demand with custom inputs. The execution table shows the steps from opening the UI, triggering the DAG with or without parameters, task execution printing the parameter, and DAG completion.

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()