🦔

Dagster Components

に公開
Dagster
tech

普段使ってるツールの使ってない機能を使ってみる Advent Calendar 2025です。

名前だけは知っていたDagster Componentsを使ってみます。

Dagster Componetns

2025年の九月(1.11.10)にGAになったDagsterの機能で、公式ドキュメントでは

YAML-based DSL for defining pipelines, a library of ready-made Components for popular tools, automatic documentation generation for your custom Components

つまり何か便利のDSLとして紹介されています。

専用のlandingページ的なものやYouTube動画も用意されており、開発チーム的に結構力を入れてそうな雰囲気です。

使ってみる

Tutorialがあるので、さっそくやってみます。

準備

とその前に何はともあれScaffoldを作ります。

> uvx create-dagster@latest project components_test2
tree components_test2
components_test2
├── README.md
├── pyproject.toml
├── src
│   └── components_test2
│       ├── __init__.py
│       ├── definitions.py
│       └── defs
│           └── __init__.py
├── tests
│   └── __init__.py
└── uv.lock

このチュートリアルではdagster-duckdbを使うのでそれもインストールしておきます。

uv add dagster-duckdb

Componentを作る

Componentを作ります。

> uv run dg scaffold component Tutorial
Creating module at: /home/notrogue/project/dagster/components_test2/src/components_test2/components
Scaffolded Dagster component at /home/notrogue/project/dagster/components_test2/src/components_test2/components/tutorial.py.

> tree
.
├── README.md
├── pyproject.toml
├── src
│   └── components_test2
│       ├── __init__.py
│       ├── __pycache__
│       │   └── __init__.cpython-313.pyc
│       ├── components
│       │   ├── __init__.py
│       │   └── tutorial.py
│       ├── definitions.py
│       └── defs
│           └── __init__.py
├── tests
│   └── __init__.py
└── uv.lock

srcの中にComoponentっぽいファイルができていますね。

Componentでは、

  • ひな形となるPythonクラス
  • ひな形にパラメータを与えるファイル
    • チュートリアルではYAMLですがPythonでも定義できます

の二つのファイルを使って、Asset（等）を定義します。

前者のファイルとしてtutorial.pyを追加します。このファイルでは、

  • パラメータをクラス変数として定義
    • この例ではduckdb_databaseとetl_steps
  • パラメータを使ったAsset
    • この例ではDuckDBにテーブル作成
  • Assetその他をまとめたDefinitionクラス

を定義しており、普通のDagsterのコードの経験ある人はなんとなく雰囲気わかると思います。

from dagster_duckdb import DuckDBResource
import dagster as dg
from dagster_duckdb import DuckDBResource

class ETL(dg.Model):
    url_path: str
    table: str


class Tutorial(dg.Component, dg.Model, dg.Resolvable):
    # The interface for the component
    duckdb_database: str
    etl_steps: list[ETL]

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        _etl_assets = []

        for etl in self.etl_steps:

            @dg.asset(
                name=etl.table,
            )
            def _table(duckdb: DuckDBResource):
                with duckdb.get_connection() as conn:
                    conn.execute(
                        f"""
                        create or replace table {etl.table} as (
                            select * from read_csv_auto('{etl.url_path}')
                        )
                        """
                    )

            _etl_assets.append(_table)

        return dg.Definitions(
            assets=_etl_assets,
            resources={"duckdb": DuckDBResource(database=self.duckdb_database)},
        )

dgコマンドでこのテンプレートを使う（空の）YAMLファイルを作ることができます。
（エディタで作ってもよいはず）

> uv run dg scaffold defs components_test2.components.tutorial.Tutorial tutorial
# できるファイル
> cat src/components_test2/defs/tutorial/defs.yaml
type: components_test2.components.tutorial.Tutorial

attributes: {}

パラメータを渡す記載を追加します。tableの部分がAssetのsuffixになります。

type: components_test2.components.tutorial.Tutorial

attributes:
  duckdb_database: /tmp/jaffle_platform.duckdb
  etl_steps:
    - url_path: https://raw.githubusercontent.com/dbt-labs/jaffle-shop-classic/refs/heads/main/seeds/raw_customers.csv
      table: customers
    - url_path: https://raw.githubusercontent.com/dbt-labs/jaffle-shop-classic/refs/heads/main/seeds/raw_orders.csv
      table: orders
    - url_path: https://raw.githubusercontent.com/dbt-labs/jaffle-shop-classic/refs/heads/main/seeds/raw_payments.csv
      table: payments

Dagsterを起動します。

> uv run dagster dev

Asset Catalogを見ると、通常のAsset同様に登録されています！当然実行も出来ます。

落ち葉拾い

Componentと通常のAssetが同居できるか

できます。例えば下のようなAsset定義をdefs/下に置くと、Componentによる定義とならんでAssset追加されます。

import dagster as dg

@dg.asset
def customers_normal() -> str:
    return "https://raw.githubusercontent.com/dbt-labs/jaffle-shop-classic/refs/heads/main/seeds/raw_customers.csv"

Asset以外も定義できるか

できます。

Componentを定義していたtutorial.pyに追加してみます。

# Tutorialクラスの続き
            _etl_job = dg.define_asset_job(
                name=f"etl_job_{etl.table}",
                selection=_etl_assets
            )
            _etl_jobs.append(_etl_job)

        return dg.Definitions(
            assets=_etl_assets,
            jobs=_etl_jobs,
            resources={"duckdb": DuckDBResource(database=self.duckdb_database)},
        )

Dagster WebUIで見るとジョブが追加されています。
（コードは省略しますがスケジュールを追加できることも確認。AssetCheck/Sensorは試してないですが流れ的に出来そうと予想）

何がうれしいの

Asset（等）定義できるという点では、Dagsterに元からある記載と同じなのですが、

  • 再利用しやすくなる
    • チーム内であれ他組織であれ
  • Pythonの記載よりも自動化/AIでの記載しやすい

とかなのかなーと思っています（landingページとか見ても）。

Discussion