如何处理 AIRFLOW 的错误（DatabricksSubmitRunOperator）答案

【问题标题】：How to handle errors for AIRFLOW (DatabricksSubmitRunOperator)如何处理 AIRFLOW 的错误（DatabricksSubmitRunOperator）
【发布时间】：2021-06-21 06:56:58
【问题描述】：

我有 Airflow dag ，它运行一系列 Databricks 笔记本。

现在我想要的是，如果笔记本出现故障？如何向此笔记本失败的用户发送邮件，但执行日期等细节很少。

有什么错误处理方式吗？

【问题讨论】：

添加你的 DAG 代码

标签： airflow azure-databricks

【解决方案1】：

BaseOperator 具有允许配置在失败时发送电子邮件的参数，因此它可供所有操作员使用：

DatabricksSubmitRunOperator(...,email_on_failure=True,
                            email='your.email@where.com')

【讨论】：

【解决方案2】：

第一步： 将email_on_failure设置为False，并使用运营商的on_failure_callback。 on_failure_callback下面介绍的函数。

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('you_email@address.com', title, body)

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

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。

【讨论】：