Manual triggers and parameters in Apache Airflow - Commands & Configuration

Introduction

Sometimes you want to start a workflow yourself and give it some information to use. Manual triggers let you do this in Airflow, so you can run tasks on demand with specific details.

When you want to run a data pipeline only after a manual check or approval.

When you need to test a workflow with different input values without changing the code.

When you want to rerun a failed task with updated parameters.

When you want to start a report generation with a chosen date range.

When you want to trigger a workflow from the Airflow UI with custom settings.

Config File - manual_trigger_dag.py

manual_trigger_dag.py

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

def print_params(**kwargs):
    param1 = kwargs['dag_run'].conf.get('param1', 'default1')
    param2 = kwargs['dag_run'].conf.get('param2', 'default2')
    print(f"Parameter 1: {param1}")
    print(f"Parameter 2: {param2}")

with DAG(
    dag_id='manual_trigger_dag',
    start_date=days_ago(1),
    schedule_interval=None,
    catchup=False
) as dag:
    task = PythonOperator(
        task_id='print_params_task',
        python_callable=print_params
    )

This DAG defines a workflow that does not run on a schedule (schedule_interval=None), so it only runs when manually triggered.

The print_params function reads parameters param1 and param2 from the manual trigger input and prints them.

The PythonOperator runs this function when the DAG is triggered.

Commands

This command manually starts the DAG named 'manual_trigger_dag' and passes parameters 'param1' and 'param2' with values 'hello' and 'world'.

Terminal

airflow dags trigger manual_trigger_dag --conf '{"param1":"hello","param2":"world"}'

Expected Output

Created <DagRun manual_trigger_dag @ 2024-06-01T12:00:00+00:00: manual__2024-06-01T12:00:00+00:00, externally triggered: True>

→

--conf - Pass JSON parameters to the DAG run

This command shows the logs of the task 'print_params_task' for the manual run started at the given execution date, so you can see the printed parameters.

Terminal

airflow tasks logs manual_trigger_dag print_params_task --execution-date 2024-06-01T12:00:00+00:00

Expected Output

Parameter 1: hello

Parameter 2: world

→

--execution-date - Specify the exact run to get logs from

Key Concept

If you remember nothing else from this pattern, remember: manual triggers let you start workflows anytime with custom input parameters.

Common Mistakes

Not setting schedule_interval to None in the DAG definition.

The DAG might run automatically on a schedule, which is not desired for manual triggers.

Always set schedule_interval=None to ensure the DAG only runs when manually triggered.

Passing parameters with incorrect JSON format in the --conf flag.

Airflow will reject the trigger or ignore parameters if JSON is invalid.

Use proper JSON syntax with double quotes and escape characters if needed.

Trying to access parameters without checking if they exist in dag_run.conf.

This causes errors if parameters are missing during manual or scheduled runs.

Use .get() with default values to safely access parameters.

Summary

Define a DAG with schedule_interval=None to allow manual triggering only.

Use the --conf flag with airflow dags trigger to pass parameters as JSON.

Access parameters inside tasks via dag_run.conf safely with default values.

Check task logs to verify parameters were received and used correctly.

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