⚡

DAG生成技能Skill dag-generate

该技能用于从 Starlake 项目配置自动生成 Airflow 或 Dagster 的 DAG（有向无环图）文件，支持数据加载和变换任务的编排，适用于数据工程、ETL 开发和自动化工作流管理。关键词：DAG, Airflow, Dagster, Starlake, 数据工程, ETL, 任务编排, 工作流。

数据工程 0 次安装 0 次浏览 更新于 3/15/2026

name: dag-generate description: 从您的 Starlake 项目生成编排 DAG（Airflow/Dagster）

DAG 生成技能

从您的 Starlake 项目配置生成编排 DAG 文件（Airflow Python DAGs 或 Dagster jobs）。DAGs 是从 Jinja2 模板生成的，并且可以按域、表或任务自定义。

用法

starlake dag-generate [options]

选项

  • --outputDir <value>: 输出目录用于生成的 DAG 文件
  • --clean: 在生成前清理输出目录
  • --tags <value>: 仅生成匹配这些标签的任务/表的 DAG
  • --tasks: 生成变换任务的 DAG 文件
  • --domains: 生成加载域的 DAG 文件
  • --withRoles: 在生成的 DAG 中包含角色定义
  • --reportFormat <value>: 报告输出格式: console, json, 或 html

配置上下文

DAG 配置文件 (metadata/dags/{name}.sl.yml)

每个 DAG 配置指定一个模板和输出文件名:

# metadata/dags/airflow_load_shell.sl.yml
version: 1
dag:
  comment: "示例 dag 配置"
  template: "load/airflow__scheduled_table__shell.py.j2"
  filename: "airflow_all_tables.py"
  options:
    sl_env_var: '{"SL_ROOT": "{{SL_ROOT}}"}'

# metadata/dags/airflow_transform_shell.sl.yml
dag:
  comment: "示例 dag 配置"
  template: "transform/airflow__scheduled_task__shell.py.j2"
  filename: "airflow_all_tasks.py"
  options:
    run_dependencies_first: "true"

Dagster DAG 配置

# metadata/dags/dagster_load_shell.sl.yml
dag:
  comment: "{{domain}} 的数据加载"
  template: "load/dagster__scheduled_table__shell.py.j2"
  filename: "dagster_all_load.py"
  options:
    run_dependencies_first: "true"
    sl_env_var: '{"SL_ROOT": "{{SL_ROOT}}"}'
    SL_STARLAKE_PATH: "{{SL_ROOT}}/starlake"
    pre_load_strategy: "none"       # 待处理, 已导入, 确认, 无
    global_ack_file_path: "{{SL_ROOT}}/datasets/pending/starbake/GO.ack"
    ack_wait_timeout: "60"          # 秒

DAG 参考分配

DAGs 在 application.sl.yml 中在项目级别分配，并且可以在域或表级别覆盖:

# metadata/application.sl.yml
application:
  dagRef:
    load: "airflow_load_shell"       # 所有加载任务的默认 DAG
    transform: "airflow_transform_shell"  # 所有变换的默认 DAG

在域级别覆盖:

# metadata/load/starbake/_config.sl.yml
load:
  metadata:
    dagRef: "custom_load_dag"

在表/任务级别覆盖:

# metadata/load/starbake/orders.sl.yml
table:
  metadata:
    dagRef: "orders_specific_dag"

优先级 (从低到高): 项目 → 域 → 表/任务

示例

生成所有 DAG

starlake dag-generate --outputDir /tmp/dags --clean

仅生成加载 DAG

starlake dag-generate --domains --outputDir /tmp/dags

仅生成变换 DAG

starlake dag-generate --tasks --outputDir /tmp/dags

为特定标签生成 DAG

starlake dag-generate --tags daily,critical --outputDir /tmp/dags

生成带角色定义的 DAG

starlake dag-generate --outputDir /tmp/dags --withRoles --clean

相关技能