dag-factory

DAG Factory

dag-factory は YAML 設定ファイルを Airflow DAG に変換するライブラリです。ユーザーが Apache Airflow DAG を宣言的に構築できるよう支援します。手順は順番に実行し、ユーザーのニーズを満たす最もシンプルな設定を優先してください。

パッケージ：PyPI の dag-factory リポジトリ：https://github.com/astronomer/dag-factory ドキュメント：https://astronomer.github.io/dag-factory/latest/ 対象：dag-factory v1.0+ のみ。v1.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 を使う」	カスタムオペレーターとプロバイダーオペレーターへ進む
「動的タスク／マッピングされたタスク」／「expand / partial」	動的タスクマッピングへ進む
「データセットでスケジュールする」／「アウトレットとインレット」	データセットへ進む
「コールバックを追加する」／「失敗時に Slack 通知する」	コールバックへ進む
「タイムテーブルを使う」／「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 プロバイダーパッケージを自動でインストールしません。YAML で参照するプロバイダーパッケージ（例：apache-airflow-providers-slack、apache-airflow-providers-cncf-kubernetes）をインストールしてください。

2. ローダーを作成する

Airflow の DAG プロセッサーが読み込めるように、dags/load_dags.py を作成します。

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={...})

生成された DAG オブジェクトが、Airflow から検出できるモジュールの名前空間に登録されるよう、globals_dict=globals() が必要です。

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 キーワード引数
tasks	DAG	タスク辞書のリスト。各辞書に task_id と 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 をキーとする辞書形式）も後方互換性のために引き続き使用できますが、新しいコードではリスト形式を推奨します。

---

デフォルト

デフォルトを設定する方法は 4 つあり、優先順位の高い順に並んでいます。

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 ブロック

1 つのファイルから複数の 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 ファイル

DAG YAML と同じ場所に defaults.yml を配置するか、親ディレクトリを 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

---

カスタムおよびプロバイダーのオペレーター

オペレーターは、完全な Python インポートパスで指定してください。dag-factory は、その他のすべてのタスクキーをそのオペレーターへのキーワード引数として渡します。

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

オペレーターのパッケージがインストールされ、インポート可能である必要があります。Airflow 3 では、レガシーな airflow.operators.* パスよりも airflow.providers.standard.operators.* を優先してください。dagfactory convert CLI はこれらのパスを自動的に書き換えます。

KubernetesPodOperator

オペレーターのパスを指定し、キーワード引数を直接渡します。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 では、上流タスクの出力を参照する方法が 2 種類あります。

• task_id.output — XCom 形式の参照です。expand の op_args / op_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 を使用してデータセットの生成元を宣言し、schedule にデータセット URI のリストを指定して消費します。

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+）

論理演算子 __and__ ／ __or__ を datasets キーの下にネストします。

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

---

コールバック

次の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. プロバイダーのコールバック

- 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"

プロバイダーパッケージをインストールする必要があります。

---

カスタム 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__。items キーも、__type__: builtins.list ブロック内で使用する場合は予約済みです。型付きリストの構築に、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 の解析時に表面化します。

---

トラブルシューティング

「Operator not found」／ModuleNotFoundError

原因：プロバイダーパッケージがインストールされていないか、インポートパスが誤っています。

解決方法：プロバイダーをインストールし（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）を確認してください。

「Argument is not JSON-serializable」／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(), ...) を呼び出している
• 必要な Airflow プロバイダーが requirements.txt に記載されている
• インポートエラーなしで DAG が Airflow UI に表示される

---

関連スキル

• authoring-dags — af CLI による検証を使って、純粋な Python で Airflow DAG を作成します。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/