Conditional Email Notification on DAG Failure

marie20 0
Viewed 23

I have a Airflow 1.10 DAG with the following sequence of operators -

PythonOperator1 --> S3KeySensor --> PythonOperator2 --> PythonOperator3

My requirement is to send email notification if -

  • S3KeySensor fails (timeout occurs waiting for file with soft_fail=True i.e. skipped)
  • PythonOperator2 or PythonOperator3 fails
  • No need to send email if DAG completes successfully

Can anyone please help how to implement this conditional logic with the EmailOperator.

Thanks.

airflow
1 Answers

There is a on_failure_callback config option for every Airflow task. It's value is supposed to be a python callable. So you'd need to configure your tasks (S3KeySensor, PythonOperator based) like that:

def send_email(context=None):
    from airflow.operators.email import EmailOperator

    email_alert = EmailOperator(...)

    slack_alert.execute(context)

with DAG(...) as dag:
    task_a = S3KeySensor(on_failure_callback=send_email, ...)
    task_b = PythonOperator(on_failure_callback=send_email, ...)

    task_a > task_b

More info on callbacks:

Remember that Airflow 1.* is deprecated already so it's recommended to use Airflow 2+.

mrp1nk0
Related