如何处理 AIRFLOW (DatabricksSubmitRunOperator) 的错误 [英] How to handle errors for AIRFLOW (DatabricksSubmitRunOperator)

问题描述

我有运行 Databricks 笔记本系列的 Airflow dag.

I have Airflow dag , which runs series of Databricks notebook.

现在我想要的是，如果笔记本出现故障?如何向此笔记本失败的用户发送邮件，但没有执行日期等详细信息.

Now What i want is , if any one the notebook got failed? How to send mail to user that this notebook got failed with few details like execution date.

有什么错误处理方法吗?

and is there any error handling way?

推荐答案

Step1: 将 email_on_failure 设置为 False 并使用操作员的 >on_failure_callback.on_failure_callback 下面描述的函数.

Step1: Set email_on_failure to False and use the operators’s on_failure_callback. on_failure_callback the function described below.

from airflow.utils.email import send_email

def notify_email(contextDict, **kwargs):
    """Send custom email alerts."""

    # email title.
    title = "Airflow alert: {task_name} Failed".format(**contextDict)

    # email contents
    body = """
    Hi Everyone, <br>
    <br>
    There's been an error in the {task_name} job.<br>
    <br>
    Forever yours,<br>
    Airflow bot <br>
    """.format(**contextDict)

    send_email('[e-mail]', title, body)

第 2 步: 下面是气流 dag 的简短示例

Step2: Short example airflow dag below

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

args = {
  'owner': 'me',
  'description': 'my_example',
  'start_date': days_ago(1)
}

# run every day at 12:05 UTC
dag = DAG(dag_id='example_dag', default_args=args, schedule_interval='0 5 * * *')

def print_hello():
  return 'hello!'

py_task = PythonOperator(task_id='example',
                         python_callable=print_hello,
                         on_failure_callback=notify_email,
                         dag=dag)

py_task

注意在 PythonOperator 中设置 on_failure_callback 等于 notify_email 的地方.

Note where set on_failure_callback equal to notify_email in the PythonOperator.

这篇关于如何处理 AIRFLOW (DatabricksSubmitRunOperator) 的错误的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！