dag-factory

DAG Factory

你正在帮助用户使用 dag-factory 以声明式方式构建 Apache Airflow DAG。dag-factory 是一个可将 YAML 配置文件转换为 Airflow DAG 的库。按顺序执行步骤，并优先采用满足用户需求的最简单配置。

软件包：PyPI 上的 dag-factory 代码仓库：https://github.com/astronomer/dag-factory 文档：https://astronomer.github.io/dag-factory/latest/ 目标版本：仅限 dag-factory v1.0+。对于 1.0 之前的项目，请先参阅 reference/migration.md，再应用本技能中的任何指导。 要求：Python 3.10+、Airflow 2.4+（支持 Airflow 3）

开始之前

与用户确认：

1. Airflow 版本 ≥2.4
2. Python 版本 ≥3.10
3. dag-factory 版本：本技能面向 v1.0+。如果项目使用 <1.0 版本，请按照 reference/migration.md 升级后再继续。
4. 使用场景：dag-factory 用于声明式、低代码 DAG 编写。如果用户需要可复用、经过验证的 Pydantic Python 风格模板，请推荐 blueprint。如果用户需要完整的 Python 灵活性，请推荐 authoring-dags 技能。

---

确定用户的需求

用户请求	操作
“创建 YAML DAG” / “将这个 Python DAG 转换为 YAML”	转到 在 YAML 中定义 DAG
“在我的项目中设置 dag-factory”	转到 项目设置
“在多个 DAG 之间共享默认值” / “只设置一次 start_date”	转到 默认值
“使用自定义运算符” / “使用 KPO / Slack / Snowflake”	转到 自定义和 Provider 运算符
“动态 / 映射任务” / “expand / partial”	转到 动态任务映射
“按数据集调度” / “Outlets 和 inlets”	转到 数据集
“添加回调” / “失败时发送 Slack 通知”	转到 回调
“使用 timetable” / “YAML 中的 datetime” / “YAML 中的 timedelta”	转到 自定义 Python 对象（__type__）
“检查我的 YAML” / “验证”	转到 验证命令
“将 Airflow 2 YAML 转换为 Airflow 3”	转到 验证命令（dagfactory convert）
“从 dag-factory <1.0 迁移”	参阅 reference/migration.md
dag-factory 报错 / 故障排除	转到 故障排除

---

项目设置

1. 安装软件包

添加到 requirements.txt：

dag-factory>=1.0.0

dag-factory 不会自动安装 Airflow providers。请安装 YAML 中引用的任何 provider 软件包（例如 apache-airflow-providers-slack、apache-airflow-providers-cncf-kubernetes）。

2. 创建加载器

创建 dags/load_dags.py，以便 Airflow 的 DAG 处理器能够加载该文件：

import os
from pathlib import Path

from dagfactory import load_yaml_dags

CONFIG_ROOT_DIR = Path(os.getenv("CONFIG_ROOT_DIR", "/usr/local/airflow/dags/"))

# Option A: load every *.yml / *.yaml under a folder
load_yaml_dags(globals_dict=globals(), dags_folder=str(CONFIG_ROOT_DIR))

# Option B: load a single file
# load_yaml_dags(globals_dict=globals(), config_filepath=str(CONFIG_ROOT_DIR / "my_dag.yml"))

# Option C: load from an in-Python dict
# load_yaml_dags(globals_dict=globals(), config_dict={...})

必须设置 globals_dict=globals()，这样生成的 DAG 对象才能注册到 Airflow 可发现的模块命名空间中。

3. 验证安装

dagfactory --version

---

在 YAML 中定义 DAG

每个顶层 YAML 键（default 除外）都会定义一个 DAG。该键会成为 dag_id。tasks 和 task_groups 请使用列表格式——这是 v1.0.0 起推荐的格式。

# dags/example_dag_factory.yml
default:
  default_args:
    start_date: 2024-11-11

basic_example_dag:
  default_args:
    owner: "custom_owner"
  description: "this is an example dag"
  schedule: "0 3 * * *"
  catchup: false
  task_groups:
    - group_name: "example_task_group"
      tooltip: "this is an example task group"
      dependencies: [task_1]
  tasks:
    - task_id: "task_1"
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo 1"
    - task_id: "task_2"
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo 2"
      dependencies: [task_1]
    - task_id: "task_3"
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo 3"
      dependencies: [task_1]
      task_group_name: "example_task_group"

关键字段

字段	位置	用途
default	顶层	应用于此文件中每个 DAG 的共享 DAG 级参数
default_args	DAG 或 default 块	标准 Airflow default_args（owner、retries、start_date 等）
schedule	DAG	Cron 表达式、预设值（@daily）、Dataset 列表或 __type__ timetable
catchup / description / tags	DAG	标准 Airflow DAG kwargs
tasks	DAG	任务字典列表；每项都必须包含 task_id 和 operator
operator	任务	完整导入路径，指向 operator 类（例如 airflow.operators.bash.BashOperator）
dependencies	任务 / task_group	上游 task_id 或 group_name 列表
task_groups	DAG	分组字典列表；每项都必须包含 group_name
task_group_name	任务	将任务分配到一个任务组

YAML 中的任务无需按依赖关系排序 — dag-factory 会解析 DAG 拓扑结构。

字典格式（旧版）

为保持向后兼容，1.0 之前的字典格式（其中 tasks 是以 task_id 为键的字典）仍然有效，但新代码建议使用列表格式。

---

默认值

有四种设置默认值的方法，按优先级顺序排列（从高到低）：

1. 单个 DAG 内的 default_args / DAG 级键
2. 同一 YAML 文件中的顶层 default: 块
3. 传给 load_yaml_dags 的 defaults_config_dict= 参数
4. 通过 defaults_config_path= 指定的 defaults.yml（或 defaults.yaml）文件（或自动检测 DAG YAML 旁的文件）

注意：加载器参数名和其他几个字段名在 v1.0.0 中发生了变化。如果你使用的是较旧项目，请参阅 reference/migration.md。

同一文件中的 default 块

适合从一个文件模板化多个 DAG：

default:
  default_args:
    owner: "data-team"
    start_date: 2025-01-01
    retries: 2
  catchup: false
  schedule: "@daily"

dag_one:
  description: "first DAG"
  tasks:
    - task_id: t1
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo one"

dag_two:
  description: "second DAG"
  tasks:
    - task_id: t1
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo two"

defaults.yml 文件

将 defaults.yml 放在 DAG YAML 旁，或将 defaults_config_path 指向父目录。dag-factory 会沿目录树向上查找并合并所有 defaults.yml 文件，离 DAG YAML 最近的文件优先。DAG 级参数（例如 schedule、catchup）放在 defaults.yml 根级；每个任务的默认值放在 default_args 下。

# defaults.yml
schedule: 0 1 * * *
catchup: false
default_args:
  start_date: '2024-12-31'
  owner: data-team

---

自定义和 Provider Operators

通过完整的 Python 导入路径引用任何 operator。dag-factory 会将其他所有任务键作为 kwargs 传给该 operator。

tasks:
  - task_id: begin
    operator: airflow.providers.standard.operators.empty.EmptyOperator
  - task_id: make_bread
    operator: customized.operators.breakfast_operators.MakeBreadOperator
    bread_type: 'Sourdough'

operator 的包必须已安装且可导入。对于 Airflow 3，建议使用 airflow.providers.standard.operators.*，而不是旧版 airflow.operators.* 路径 — dagfactory convert CLI 会自动重写这些路径。

KubernetesPodOperator

指定 operator 路径并直接传入 kwargs。从 v1.0 起，dag-factory 不再执行旧版类型转换 — 嵌套的 k8s 对象请使用 __type__。

tasks:
  - task_id: hello-world-pod
    operator: airflow.providers.cncf.kubernetes.operators.pod.KubernetesPodOperator
    image: "python:3.12-slim"
    cmds: ["python", "-c"]
    arguments: ["print('hi')"]
    name: example-pod
    namespace: default
    container_resources:
      __type__: kubernetes.client.models.V1ResourceRequirements
      limits: {cpu: "1", memory: "1024Mi"}
      requests: {cpu: "0.5", memory: "512Mi"}

---

动态任务映射

在任务上使用 expand 和 partial 键进行动态映射。dag-factory 提供两种不同方式来引用上游任务的输出：

• task_id.output — XCom 风格的引用，用于 expand 的 op_args / op_kwargs 内（以及其他 operator 中的等效 kwargs）。
• +task_id — 裸值引用，用于值直接位于 expand 下的情况（例如 expand: {number: +numbers_list}），或作为 TaskFlow 装饰器参数。

不要混用：+request 在 op_args 内无法解析，而 request.output 不能作为裸 expand 值解析。

dynamic_task_map:
  default_args:
    start_date: 2025-01-01
  schedule: "0 3 * * *"
  tasks:
    - task_id: request
      operator: airflow.providers.standard.operators.python.PythonOperator
      python_callable_name: make_list
      python_callable_file: $CONFIG_ROOT_DIR/expand_tasks.py
    - task_id: process
      operator: airflow.providers.standard.operators.python.PythonOperator
      python_callable_name: consume_value
      python_callable_file: $CONFIG_ROOT_DIR/expand_tasks.py
      partial:
        op_kwargs:
          fixed_param: "test"
      expand:
        op_args: request.output    # XCom-style — used inside op_args / op_kwargs
      dependencies: [request]

裸值形式（TaskFlow decorator 任务，或任何非 op_args 映射）：

tasks:
  - task_id: numbers_list
    decorator: airflow.sdk.definitions.decorators.task
    python_callable: sample.build_numbers_list
  - task_id: double_number
    decorator: airflow.sdk.definitions.decorators.task
    python_callable: sample.double
    expand:
      number: +numbers_list   # + resolves to upstream task `numbers_list`'s XComArg

对于具名映射索引（Airflow 2.9+），设置 map_index_template: "{{ task.custom_mapping_key }}"，并让可调用对象赋值 context["custom_mapping_key"]。

已测试模式：简单映射、任务生成的映射、重复映射、partial、多参数映射、map_index_template。 不支持 / 未测试：对任务组进行映射、压缩、转换扩展数据。

---

数据集

在任务中使用 inlets / outlets 声明数据集生产者，并使用数据集 URI 列表作为 schedule 来消费数据集。

producer_dag:
  default_args:
    start_date: '2024-01-01'
  schedule: "0 5 * * *"
  catchup: false
  tasks:
    - task_id: task_1
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo 1"
      outlets: ['s3://bucket_example/raw/dataset1.json']

consumer_dag:
  default_args:
    start_date: '2024-01-01'
  schedule: ['s3://bucket_example/raw/dataset1.json']
  catchup: false
  tasks:
    - task_id: task_1
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo 'consumer'"

条件数据集调度（Airflow 2.9+ / dag-factory 0.22+）

在 datasets 键下嵌套逻辑运算符 __and__ / __or__。

schedule:
  datasets:
    __or__:
      - __and__:
          - s3://bucket-cjmm/raw/dataset_custom_1
          - s3://bucket-cjmm/raw/dataset_custom_2
      - s3://bucket-cjmm/raw/dataset_custom_3

---

回调

三种方式均有效，可用于 DAG、TaskGroup 或任务级别（也可放在 default_args 下）：

1. 指向可调用对象的字符串

- task_id: task_1
  operator: airflow.operators.bash.BashOperator
  bash_command: "echo task_1"
  on_failure_callback: include.custom_callbacks.output_standard_message

带 kwargs：

- task_id: task_2
  operator: airflow.operators.bash.BashOperator
  bash_command: "echo task_2"
  on_success_callback:
    callback: include.custom_callbacks.output_custom_message
    param1: "Task status"
    param2: "Successful!"

2. 文件路径 + 函数名（不带 kwargs）

- task_id: task_3
  operator: airflow.operators.bash.BashOperator
  bash_command: "echo task_3"
  on_retry_callback_name: output_standard_message
  on_retry_callback_file: /usr/local/airflow/include/custom_callbacks.py

3. Provider 回调

- task_id: task_4
  operator: airflow.operators.bash.BashOperator
  bash_command: "echo task_4"
  on_failure_callback:
    callback: airflow.providers.slack.notifications.slack.send_slack_notification
    slack_conn_id: slack_conn_id
    text: ":red_circle: Task Failed."
    channel: "#channel"

必须安装 Provider 包。

---

自定义 Python 对象（__type__）

对于非简单标量的内容——datetime、timedelta、Asset、时间表、k8s 对象——请使用通用对象语法：

start_date:
  __type__: datetime.datetime
  year: 2025
  month: 1
  day: 1

execution_timeout:
  __type__: datetime.timedelta
  hours: 1

schedule:
  __type__: airflow.timetables.trigger.CronTriggerTimetable
  cron: "0 1 * * 3"
  timezone: UTC

• __type__ 是类的完整导入路径
• __args__ 是位置参数列表
• 其他键会成为关键字参数
• 对于类型化对象列表，请使用 __type__: builtins.list，并搭配 items: 键

保留键

不要将以下 YAML 键用于自定义数据——dag-factory 将其保留：__type__、__args__、__join__、__and__、__or__。在 __type__: builtins.list 块中使用时，items 也属于保留键——不要在类型化列表构造中添加名为 items 的自定义字段。

---

验证命令

安装后，dagfactory CLI 可在 PATH 中使用：

命令	使用时机
dagfactory --version	确认安装状态 / 版本
dagfactory lint <path>	验证文件或目录的 YAML 语法
dagfactory lint <path> --verbose	显示逐文件结果表
dagfactory convert <path>	显示将 Airflow 2 → 3 导入路径迁移所需的差异
dagfactory convert <path> --override	原地应用转换

验证工作流

# 1. Lint YAML
dagfactory lint dags/

# 2. Have Airflow parse to catch operator/import errors
#    (Astro CLI users)
astro dev parse

dagfactory lint 只检查 YAML 语法——运算符导入错误和缺失的 kwargs 会在 Airflow 解析时显现。

---

故障排除

“找不到运算符” / ModuleNotFoundError

原因：未安装 Provider 包，或导入路径错误。

修复：安装 Provider（pip install apache-airflow-providers-...）并验证路径。对于 Airflow 3，运行 dagfactory convert，将旧版 airflow.operators.* 路径更新为 airflow.providers.standard.operators.*。

YAML 解析成功，但 DAG 未在 Airflow 中显示

原因：缺少加载器文件，或未传入 globals_dict=globals()。

修复：确保 dags/ 中的 Python 文件调用 load_yaml_dags(globals_dict=globals(), ...)。检查 astro dev parse（或 airflow dags list-import-errors）中的解析错误。

“参数无法进行 JSON 序列化” / kwargs 类型错误

原因：需要 Python 对象时却传入了标量字符串（例如，字段需要 datetime 时使用 start_date: "2025-01-01"）。

修复：按照自定义 Python 对象一节中的说明，使用 __type__: datetime.datetime（或 datetime.timedelta 等）。

条件数据集调度被忽略

原因：Airflow <2.9、dag-factory <0.22，或使用了旧版 !and / !or 键。

修复：升级并重命名为 __and__ / __or__。

多个 defaults.yml 未按预期合并

原因：defaults_config_path 未指向 DAG YAML 的父目录。

修复：将 defaults_config_path 设置为你希望纳入的最高层级祖先目录；dag-factory 会从 DAG 文件向上遍历目录树并按该顺序合并，离 DAG 越近的文件优先级越高。

---

验证清单

完成前，请与用户确认：

• dagfactory lint dags/ 运行通过
• dags/ 中存在加载器文件，且调用了 load_yaml_dags(globals_dict=globals(), ...)
• requirements.txt 中包含所需的 Airflow Provider
• DAG 显示在 Airflow UI 中，且没有导入错误

---

相关技能

• authoring-dags — 使用纯 Python 编写 Airflow DAG，并通过 af CLI 验证。适用于 YAML 无法满足需求的情况。
• testing-dags：用于测试 DAG、调试故障，以及执行测试 -> 修复 -> 重新测试循环
• debugging-dags：用于排查失败的 DAG

参考

• GitHub：https://github.com/astronomer/dag-factory
• 文档：https://astronomer.github.io/dag-factory/latest/
• PyPI：https://pypi.org/project/dag-factory/
• 迁移指南：https://astronomer.github.io/dag-factory/latest/migration_guide/