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.