Task failure callbacks in Apache Airflow - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to define a task failure callback function in Airflow.

Apache Airflow

def task_failure_callback(context):
    print('Task failed:', context[1])

Drag options to blanks, or click blank then click option'

A['task_instance'].task_id

B['task_instance']

C['dag']

D['execution_date']

Attempts:

3 left

💡 Hint

Common Mistakes

Using context['task_instance'] directly without accessing task_id.

Using incorrect keys like 'dag' or 'execution_date' for task ID.

The context dictionary contains details about the task instance. Accessing context['task_instance'].task_id prints the failed task's ID.

2fill in blank

medium

Complete the code to assign the failure callback to a task in Airflow.

Apache Airflow

task = PythonOperator(
    task_id='my_task',
    python_callable=my_function,
    on_failure_callback=[1]
)

Drag options to blanks, or click blank then click option'

Aprint

Btask_failure_callback()

Cmy_function

Dtask_failure_callback

Attempts:

3 left

💡 Hint

Common Mistakes

Adding parentheses after the callback function name.

Passing unrelated functions like 'my_function' or 'print'.

The on_failure_callback parameter expects a function reference, not a function call. So, pass task_failure_callback without parentheses.

3fill in blank

hard

Fix the error in the failure callback function to correctly log the task ID.

Apache Airflow

def task_failure_callback(context):
    task_id = context[1]
    print(f'Task failed: {task_id}')

Drag options to blanks, or click blank then click option'

A['task_instance']

B['task_id']

C['task_instance'].task_id

D['dag'].task_id

Attempts:

3 left

💡 Hint

Common Mistakes

Using context['task_id'] which does not exist.

Not accessing the task_id attribute of task_instance.

The task ID is accessed via context['task_instance'].task_id. Using just context['task_id'] or context['task_instance'] alone will not give the task ID string.

4fill in blank

hard

Fill both blanks to create a failure callback that sends an email with the failed task's ID.

Apache Airflow

def task_failure_callback(context):
    task_id = context[1]
    send_email(to='[e-mail]', subject=f'Task [2] Failed', body=f'Task {task_id} failed.')

Drag options to blanks, or click blank then click option'

A['task_instance'].task_id

Btask_id

D['dag'].dag_id

Attempts:

3 left

💡 Hint

Common Mistakes

Using context keys incorrectly.

Not using the task_id variable in the email subject.

The first blank gets the task ID from the context. The second blank uses the variable task_id in the email subject to show which task failed.

5fill in blank

hard

Fill all three blanks to define a failure callback that logs the task ID, execution date, and DAG ID.

Apache Airflow

def task_failure_callback(context):
    task_id = context[1]
    exec_date = context[2]
    dag_id = context[3]
    print(f'Task {task_id} failed on {exec_date} in DAG {dag_id}')

Drag options to blanks, or click blank then click option'

A['task_instance'].task_id

B['execution_date']

C['dag'].dag_id

D['task_instance']

Attempts:

3 left

💡 Hint

Common Mistakes

Using context keys without accessing attributes.

Mixing up keys like 'dag' and 'task_instance'.

Access the task ID via context['task_instance'].task_id, the execution date via context['execution_date'], and the DAG ID via context['dag'].dag_id.

Practice

1. What is the main purpose of on_failure_callback in an Airflow task?

easy

A. To run a function automatically when the task fails

B. To schedule the task to run at a specific time

C. To retry the task multiple times on failure

D. To log the task's successful completion

Solution

1. Step 1: Understand the role of on_failure_callback

   This parameter is designed to specify a function that runs automatically when a task fails.

2. Step 2: Differentiate from other task features

   Scheduling and retries are handled by other parameters, not on_failure_callback.

3. Final Answer:

   To run a function automatically when the task fails -> Option A

4. Quick Check:

   Failure callback = automatic function on failure [OK]

Hint: Failure callback runs code only when task fails [OK]

Common Mistakes:

• Confusing failure callback with scheduling
• Thinking it retries the task automatically
• Assuming it runs on task success

2. Which of the following is the correct way to assign a failure callback function named notify_failure to an Airflow task?

easy

A. task = PythonOperator(failure_callback=notify_failure, ...)

B. task = PythonOperator(on_failure_callback=notify_failure, ...)

C. task = PythonOperator(on_fail_callback=notify_failure, ...)

D. task = PythonOperator(callback_on_failure=notify_failure, ...)

Solution

1. Step 1: Recall the exact parameter name

   The correct parameter to assign a failure callback is on_failure_callback.

2. Step 2: Match the parameter with the options

   Only task = PythonOperator(on_failure_callback=notify_failure, ...) uses on_failure_callback correctly.

3. Final Answer:

   task = PythonOperator(on_failure_callback=notify_failure, ...) -> Option B

4. Quick Check:

   Correct parameter name = on_failure_callback [OK]

Hint: Use exact parameter name: on_failure_callback [OK]

Common Mistakes:

• Using incorrect parameter names like failure_callback
• Swapping words in the parameter name
• Misspelling the parameter

3. Given this code snippet, what will be printed if the task fails?

def alert(context):
    print(f"Task {context['task_instance'].task_id} failed")

task = PythonOperator(
    task_id='my_task',
    python_callable=lambda: 1/0,
    on_failure_callback=alert
)

medium

A. Task succeeded

B. ZeroDivisionError

C. No output, task succeeds

D. Task my_task failed

Solution

1. Step 1: Analyze the task's python_callable

   The lambda divides 1 by 0, causing a failure (ZeroDivisionError).

2. Step 2: Understand the failure callback behavior

   On failure, the alert function runs and prints the failure message with the task ID.

3. Final Answer:

   Task my_task failed -> Option D

4. Quick Check:

   Failure triggers alert print [OK]

Hint: Failure triggers callback print message [OK]

Common Mistakes:

• Expecting ZeroDivisionError to print directly
• Thinking task succeeds despite error
• Ignoring the callback print output

4. You wrote this failure callback function:

def fail_alert():
    print("Task failed")

task = PythonOperator(
    task_id='fail_task',
    python_callable=lambda: 1/0,
    on_failure_callback=fail_alert
)

Why does the callback not run when the task fails?

medium

A. The callback function must be decorated with @task

B. The callback function must return a value

C. The callback function must accept a context parameter

D. The callback function must be async

Solution

1. Step 1: Check callback function signature

   Failure callbacks receive a context dictionary argument; missing it causes errors.

2. Step 2: Understand Airflow callback requirements

   The callback must accept one parameter (context) to work properly.

3. Final Answer:

   The callback function must accept a context parameter -> Option C

4. Quick Check:

   Callback needs context argument [OK]

Hint: Callback must accept context argument [OK]

Common Mistakes:

• Defining callback without parameters
• Expecting return value from callback
• Making callback async unnecessarily

5. You want to send an email alert only when a task fails and the failure reason contains the word "timeout". How should you implement the on_failure_callback function?

hard

A. Define a function that checks context['exception'] for 'timeout' and sends email if true

B. Define a function that always sends email regardless of failure reason

C. Use on_failure_callback to retry the task on timeout

D. Set on_failure_callback to None and handle alerts elsewhere

Solution

1. Step 1: Understand the requirement

   Send email only if failure reason contains "timeout".

2. Step 2: Use context to check exception message

   The failure callback receives context with exception info; check if 'timeout' is in the exception string.

3. Step 3: Send email conditionally

   Send email alert only if the condition is met inside the callback function.

4. Final Answer:

   Define a function that checks context['exception'] for 'timeout' and sends email if true -> Option A

5. Quick Check:

   Check exception text in callback before alert [OK]

Hint: Check exception text in callback before alert [OK]

Common Mistakes:

• Sending email on all failures without filtering
• Trying to retry task in failure callback
• Ignoring exception details in context