从 Airflow 中的另一个 DAG 获取 DAG 电子邮件答案

【问题标题】：Get DAG Email from another DAG in Airflow从 Airflow 中的另一个 DAG 获取 DAG 电子邮件
【发布时间】：2022-06-11 04:36:39
【问题描述】：

我想使用气流中的另一个 DAG 获取此 DAG 的默认参数中提到的电子邮件。我怎样才能做到这一点？请帮忙，我是气流新手！

from airflow.models import DagRun
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
from datetime import datetime, timedelta
from airflow import DAG

def first_function(**context):
    print("hello")

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['example@gmail.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    'main',
    default_args=default_args,
    description='Sample DAG',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2022,6,10),
    catchup=False
) as dag:
    first_function = PythonOperator(
        task_id="first_function",
        python_callable=first_function,
    )
first_function

【问题讨论】：

标签： python airflow airflow-scheduler directed-acyclic-graphs

【解决方案1】：

您可以在 Airflow 中使用 custom module 在 DAG 之间共享配置选项/运算符/或任意 Python 代码。

通常您会在 DAG 目录中创建一个目录（默认为 {AIRFLOW_HOME}/dags。

要在 DAG 之间共享 default_args，您可以创建以下布局：

创建{AIRFLOW_HOME}/dags/custom/__init__.py
创建{AIRFLOW_HOME}/dags/custom/shared_config.py
创建{AIRFLOW_HOME}/dags/.airflowignore
将目录名称custom添加到.airflowignore的第一行。
将 default_args 字典从 DAG 剪切并粘贴到 {AIRFLOW_HOME}/dags/custom/shared_config.py

您可以在 Airflow 文档here 中看到建议的这种布局。

.airflowignore 告诉调度程序在解析 DAG 时跳过 custom 目录（默认情况下每 30 秒发生一次） - 因为 custom 目录包含 Python，但绝不包含任何 DAG，调度程序应该跳过这些文件以避免给调度程序增加不必要的负载。这在上面的文档链接中进行了解释。

您需要将 __init__.py 添加到自定义目录 - 即使在 Python3 中编写时，由于隐式命名空间您不需要它（同样在上面的同一链接中对此进行了解释），但气流需要它。

然后您可以根据需要从您的 dag 导入：

from custom.shared_config import default_args

【讨论】：