チュートリアル 14 分で読めます中級

Dagster 完全ガイド 2026：Pythonアセットベースオーケストレーションで宣言的データパイプラインを構築する

Dagsterは「ソフトウェア定義アセット」を中心に据えた宣言的なデータオーケストレーターです。Airflowとの違い、パーティションとバックフィル、dbt統合、アセットチェックまで、2026年時点の最新機能を実践的なPythonコードで解説します。

Tomás Oliveiraレビュー者 Elena Vasquez, Sep 2026
Dagster完全ガイド 2026｜Python実践

更新日: 2026年8月30日

Dagster（ダグスター）は、Pythonで宣言的にデータパイプラインを記述するためのオーケストレーターで、「ソフトウェア定義アセット（Software-Defined Assets、SDA）」を中核に据え、生成されるテーブルやモデルそのものを一級市民として扱う点でAirflowなどタスク中心の従来型ツールとは明確に違います。正直に言うと、私はFastAPIのバックエンドからETLパイプライン側に手を伸ばしたクチで、Dagsterに移行してからスケジュール依存の泥沼をようやく抜け出せました。本ガイドでは2026年時点（Dagster 1.11系）の実装を、環境構築からdbt統合、アセットチェック、本番運用まで一気通貫で扱います。

  • Dagsterは「タスク」ではなく「アセット」を中心にパイプラインを設計する宣言的オーケストレーターで、`@asset`デコレータで依存グラフを自動構築します。
  • Airflowとの最大の違いは、Dagsterはパイプラインが生成するデータ資産（テーブル、モデル、ファイル）そのものをコード化する点にあります。
  • パーティション定義とバックフィルを組み合わせることで、日次・時間別の再実行やギャップ埋めが宣言的に扱えます。
  • アセットチェック（Asset Checks）を使えば、Panderaと同様のスキーマ検証をパイプラインの一部として組み込めます。
  • dbt-coreとの統合は`@dbt_assets`により1行に近い設定で完了し、dbtモデル1つ1つがDagster側のアセットとして可視化されます。
  • Dagster+（旧Dagster Cloud）はServerless実行、Branch Deployment、Insightsなどの本番機能を提供し、OSS版は自前でDaemonとWebserverを運用します。

Dagsterとは何か：ソフトウェア定義アセットの思想

Dagsterはデータオーケストレーションフレームワークですが、AirflowやLuigiと決定的に異なるのは、パイプラインを「タスクの依存グラフ」ではなく「生成されるアセット（テーブル、モデル、ファイル、ダッシュボード）の依存グラフ」として捉える点にあります。ここでいうアセットとは、実行された結果として永続化される具体的なデータのことで、`s3://raw/orders.parquet`や`analytics.dbt.customer_metrics`テーブルなどが該当します。

この思想の帰結として、Dagsterでは「何を実行するか」ではなく「何を存在させたいか」を宣言します。関数に`@asset`デコレータを付ければ、それだけで依存関係付きのグラフができあがり、上流アセットの再実行時には下流のみを差分実行できます。従来のタスクベースのDAGでは「実行順序」しか表現できませんでしたが、アセットベースの世界では「データリネージ」がそのままオーケストレーションの単位になります。

この違いは実務で本当に効いてきます。私はFastAPIバックエンドとMLパイプラインを接続する仕事をしていた頃、「毎朝X時にジョブを流す」というAirflow思考で疲弊していました。Dagsterに乗り換えてからは、「このアセットを最新化するには何を再実行すべきか」をUIで直接クリックできるようになり、オンコール時の対応時間が体感で半分になったのを覚えています。公式ドキュメントのSoftware-Defined Assetsセクションは最初に読んでおくと理解が早いです。

DagsterとAirflowの違いは何か

「DagsterとAirflow、どっちを選ぶべきか」は日本のPythonデータエンジニア界隈でも頻繁に議論されるテーマです。結論から先に言えば、既存のAirflow資産を捨てられない現場はAirflow継続、新規に組むならDagsterが2026年時点のデファクト候補だと感じます。両者の位置づけを表で整理してみましょう。

比較項目Dagster 1.11Apache Airflow 3.x
設計思想アセット中心（宣言的）タスク中心（命令的）
データリネージファーストクラス、UIに標準搭載OpenLineage連携が別途必要
型システムPythonの型ヒントを`@asset`が尊重基本は文字列ベースのXCom
ローカル開発`dagster dev`で完結（SQLiteベース）PostgreSQL＋Scheduler＋Webserverが必要
データ品質検証Asset Checks標準搭載Great Expectations等の外部連携
dbt連携`@dbt_assets`でモデル単位に自動展開Cosmosなどのプラグインが必要
学習曲線アセット概念の理解が最初の山Operator/Sensorのカタログを覚える
エコシステム成熟度2026年時点で急成長中10年以上の実績、プロバイダー数が最大

Airflowが「時刻ベースのcronの上位互換」として設計されたのに対し、Dagsterは「dbtやSparkなどのモダンデータスタックが吐き出す成果物を統一的にオーケストレーションする」ことを目的に設計されました。SQLとPythonが半々の現場では、DagsterのモデルとSQLのモデルを対称に扱える設計が効いてきます。特にSQLMeshのPython×SQLデータ変換とDagsterを組み合わせると、変換ロジックはSQLMeshに委譲しつつ、実行スケジューリングと監視をDagsterに寄せる構成が現実的です。

環境構築とプロジェクト初期化

Dagsterはpip install dagster dagster-webserverだけで最小構成が動いてしまいます（本当に一瞬です）。本番相当の環境を組む場合はDaemon（dagster-daemon）とWebserverを分離し、メタデータストアとしてPostgreSQLを推奨しますが、ローカル開発では組み込みSQLiteで十分です。以下は2026年8月時点の推奨手順。

# uv（Rust製の高速パッケージマネージャ）で仮想環境を作る
uv venv .venv
source .venv/bin/activate

# 主要パッケージのインストール
uv pip install \
  "dagster>=1.11" \
  "dagster-webserver>=1.11" \
  "dagster-duckdb" \
  "dagster-dbt" \
  "dagster-polars"

# プロジェクトの雛形を生成
dagster project scaffold --name analytics_pipeline
cd analytics_pipeline

雛形はanalytics_pipeline/配下にdefs/（アセット定義）とtests/を作ります。ここで肝心なのはDefinitionsオブジェクトです。Dagsterは1つのDefinitionsにすべてのアセット、リソース、スケジュール、センサーを束ねる設計になっており、これがCLIとUIに読み込まれるエントリポイントになります。

# analytics_pipeline/definitions.py
from dagster import Definitions, load_assets_from_modules
from analytics_pipeline import assets

all_assets = load_assets_from_modules([assets])

defs = Definitions(
    assets=all_assets,
)

この状態でdagster devを実行すると、Dagster公式のGetting Startedガイドにあるとおりhttp://localhost:3000でUIが立ち上がり、アセットグラフを即座に確認できます。ローカル開発ではPostgreSQLもRedisも不要で、この体験の軽さがAirflowから移行する際の大きなインセンティブになります。

Software-Defined Assetsを書く

Software-Defined Assets（SDA）は、Dagsterの中心概念です。関数に@assetを付けるだけで、その関数の戻り値が「アセット」として扱われ、関数の引数名が「上流アセット」として自動的に依存関係になります。以下は生の注文データを取り込み、正規化した後にPolarsで集計する典型的な3層構成です。

# analytics_pipeline/assets.py
import polars as pl
from dagster import asset, MaterializeResult, MetadataValue


@asset(group_name="raw", compute_kind="python")
def raw_orders() -> pl.DataFrame:
    # 外部APIから注文データを取得する上流アセット
    # 本番では requests / httpx / boto3 で外部から取得
    return pl.DataFrame({
        "order_id": [1, 2, 3, 4],
        "customer_id": [10, 10, 20, 30],
        "amount_jpy": [1200, 3400, 890, 5600],
    })


@asset(group_name="staging", compute_kind="polars")
def stg_orders(raw_orders: pl.DataFrame) -> pl.DataFrame:
    # カラム名を統一し、金額を整数化するステージング層
    return raw_orders.with_columns(
        pl.col("amount_jpy").cast(pl.Int64).alias("amount"),
    ).drop("amount_jpy")


@asset(group_name="marts", compute_kind="polars")
def customer_ltv(stg_orders: pl.DataFrame) -> MaterializeResult:
    # 顧客ごとのLTVを集計するデータマート
    ltv = (
        stg_orders
        .group_by("customer_id")
        .agg(pl.col("amount").sum().alias("lifetime_value"))
        .sort("lifetime_value", descending=True)
    )
    return MaterializeResult(
        metadata={
            "row_count": len(ltv),
            "top_customer": MetadataValue.int(int(ltv["customer_id"][0])),
            "preview": MetadataValue.md(ltv.head(5).to_pandas().to_markdown()),
        }
    )

ここでのポイントは3つあります。第一に、stg_ordersの引数raw_ordersという名前がそのまま「上流アセットへの参照」になり、依存関係グラフを暗黙的に構築します。文字列ベースのset_upstream()を書く必要はありません。第二に、MaterializeResultのメタデータでUI上に行数や上位顧客をリッチに表示できます。第三に、compute_kindを指定するとUI上にライブラリのバッジが出るため、SQLとPythonが混在するパイプラインでも視認性が保てます。

スキーマ検証を厳密にやりたい場合は、Panderaによるスキーマ検証と組み合わせるのが定石です。Panderaでdataframeモデルを定義し、アセット関数の冒頭でschema.validate(df)を呼ぶだけで、失敗時にDagsterの実行履歴に例外が記録されます。

リソースとIO Manager

アセット関数はビジネスロジックだけを書き、外部システムへの接続はResourceとして注入します。これはFastAPIのDependency Injectionと同じ発想で、テスト時にモックへ差し替えられる利点があります。IO Managerはさらに一歩踏み込み、「アセットの戻り値をどこにどう永続化するか」を抽象化します。

from dagster import ConfigurableResource, asset
from dagster_duckdb_polars import DuckDBPolarsIOManager
from dagster import Definitions
import polars as pl


class NotionAPIResource(ConfigurableResource):
    # 外部API接続を pydantic v2 ベースで型安全に設定
    api_key: str
    base_url: str = "https://api.notion.com/v1"

    def fetch_pages(self) -> list[dict]:
        # 実際は httpx.get(...) を書く
        return [{"id": "page-1", "title": "Q3 KPI"}]


@asset(io_manager_key="duckdb_io")
def notion_pages(notion: NotionAPIResource) -> pl.DataFrame:
    return pl.DataFrame(notion.fetch_pages())


defs = Definitions(
    assets=[notion_pages],
    resources={
        "notion": NotionAPIResource(api_key={"env": "NOTION_API_KEY"}),
        "duckdb_io": DuckDBPolarsIOManager(database="warehouse.duckdb", schema="raw"),
    },
)

DuckDBPolarsIOManagerは、PolarsのDataFrameを自動的にDuckDBのテーブルへ永続化してくれます。下流アセットでnotion_pages: pl.DataFrameを引数に取ると、DuckDBからPolarsに逆変換して渡してくれるため、アセット関数はストレージの詳細を一切知らなくて済みます。DuckDB×Pythonの実践ガイドと組み合わせれば、単一ファイルで完結する分析データ基盤が組めます。

パーティションとバックフィル

パーティションは、日次・時間別・地域別など、同じアセットを複数の切り口で並列に物質化するための仕組みです。パーティションを定義しておくと、UIから「2026年8月1日～8月20日を再実行」というバックフィルが数クリックで発火できます。これはAirflowのcatchup=Trueより遥かに使いやすい設計です。

from dagster import (
    asset,
    DailyPartitionsDefinition,
    AssetExecutionContext,
)
import polars as pl

daily_partitions = DailyPartitionsDefinition(start_date="2026-01-01")


@asset(partitions_def=daily_partitions, compute_kind="python")
def daily_events(context: AssetExecutionContext) -> pl.DataFrame:
    # 指定日のイベントログを取得
    partition_date = context.partition_key  # e.g. "2026-08-30"
    context.log.info(f"取得対象日: {partition_date}")

    # S3など外部ストレージからその日のデータをロード
    return pl.read_parquet(f"s3://events/dt={partition_date}/*.parquet")


@asset(partitions_def=daily_partitions)
def daily_active_users(daily_events: pl.DataFrame) -> int:
    return daily_events["user_id"].n_unique()

UIの「Backfill」ボタンから期間を指定すると、Dagster Daemonがランを次々投入します。パーティションをまたぐ集計（月次から日次を参照するなど）にはマルチパーティションマッピングの機能があり、時間軸だけでなく「地域×日付」の直積も表現可能です。

スケジュールとセンサー

Dagsterの実行トリガーはSchedule（時刻ベース）とSensor（イベントベース）の2種類です。Sensorは典型的にはS3にオブジェクトが到着したときや、上流ジョブが完了したときに発火します。個人的には「センサーで起動されるアセットジョブ」を多用していて、cronから解放される快感は一度味わうと戻れません（本当に）。

from dagster import (
    schedule,
    sensor,
    RunRequest,
    SkipReason,
    define_asset_job,
)

daily_job = define_asset_job(
    name="daily_pipeline",
    selection=["daily_events", "daily_active_users"],
)


@schedule(cron_schedule="15 2 * * *", job=daily_job, execution_timezone="Asia/Tokyo")
def daily_2am_schedule(context):
    return RunRequest(
        partition_key=context.scheduled_execution_time.strftime("%Y-%m-%d"),
    )


@sensor(job=daily_job, minimum_interval_seconds=60)
def s3_arrival_sensor(context):
    latest_key = _peek_latest_s3_object()  # 別途実装
    if latest_key == context.cursor:
        return SkipReason("新規オブジェクトなし")

    context.update_cursor(latest_key)
    return RunRequest(
        run_key=latest_key,  # 同じキーでの重複実行を防ぐ
    )

run_keyで冪等性を担保できるのがセンサーの美点です。同じキーでRunRequestを返しても、Daemonが「既に処理済み」と判断してくれます。この設計は分散システム的にも堅牢で、私はFastAPIで書くWebhookエンドポイントと同じ考え方で処理できるため、認知負荷が低いと感じます。

dbtとの統合（dbt-core連携）

2026年時点のDagsterで最も強力な機能の1つが、dbt-coreとのネイティブ統合です。@dbt_assetsデコレータを使うと、dbtプロジェクト内のモデル1つ1つが自動的にDagsterのアセットとしてグラフに現れます。これによりPythonアセットとdbtモデルが対称に扱えるようになります。

from pathlib import Path
from dagster import AssetExecutionContext
from dagster_dbt import DbtCliResource, dbt_assets, DbtProject

DBT_PROJECT_DIR = Path(__file__).joinpath("..", "..", "dbt_project").resolve()

dbt_project = DbtProject(project_dir=DBT_PROJECT_DIR)
dbt_project.prepare_if_dev()


@dbt_assets(manifest=dbt_project.manifest_path)
def analytics_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()


defs = Definitions(
    assets=[analytics_dbt_assets, ...],
    resources={"dbt": DbtCliResource(project_dir=dbt_project)},
)

Pythonアセットの下流にdbtモデルを繋げる、あるいはdbtモデルを上流としてPythonアセット（scikit-learnの学習ジョブなど）を回す構成が、宣言的に書けます。MLflow 3の実験管理と組み合わせると、「dbtで作った特徴量テーブル→scikit-learnでの学習→MLflowにモデル登録」までを1つのアセットグラフで表現できます。

アセットチェックによるデータ品質検証

Dagster 1.6以降で導入されたAsset Checksは、パイプラインの実行結果に対して品質チェックを走らせる仕組みです。Great ExpectationsやPanderaのような外部ツールと役割は似ていますが、Dagster内で完結しUIに結果が可視化される点が強みです。

from dagster import asset, asset_check, AssetCheckResult
import polars as pl


@asset
def customer_ltv() -> pl.DataFrame:
    return pl.DataFrame({
        "customer_id": [10, 20, 30],
        "lifetime_value": [4600, 890, 5600],
    })


@asset_check(asset=customer_ltv)
def no_negative_ltv(customer_ltv: pl.DataFrame) -> AssetCheckResult:
    negatives = customer_ltv.filter(pl.col("lifetime_value") < 0)
    return AssetCheckResult(
        passed=len(negatives) == 0,
        metadata={"negative_row_count": len(negatives)},
    )


@asset_check(asset=customer_ltv, blocking=True)
def unique_customer_ids(customer_ltv: pl.DataFrame) -> AssetCheckResult:
    duplicates = (
        customer_ltv.group_by("customer_id").agg(pl.len()).filter(pl.col("len") > 1)
    )
    return AssetCheckResult(
        passed=len(duplicates) == 0,
        severity="ERROR",
    )

blocking=Trueを付けたチェックは失敗時に下流アセットの実行を止められます。私はこれを「本番投入前の関所」として使い、テストDBでは緩く、本番では厳格に、というふうにResource側で挙動を切り替えています。

Dagster+とOSS版の比較

Dagster+（旧Dagster Cloud）は、Elementl社が提供するマネージド版で、Serverless実行、Branch Deployment、Insights、Alertsなどの本番向け機能を追加します。OSSと同じPythonコードで動作しますが、運用負荷が大きく変わります。

機能Dagster OSSDagster+
ライセンスApache 2.0（無料）従量課金（実行分単位）
Webserver / Daemon自前で運用マネージド
Serverless実行非対応対応（コンテナ即起動）
Branch Deployment非対応PRごとに独立環境が立つ
Insights（実行コスト分析）非対応アセット単位で計測
SSO / RBAC非対応SAML / SCIM対応
推奨用途個人・小規模チームチーム開発・エンタープライズ

私の経験則では、5名以下のデータチームでKubernetesを触りたくない場合はDagster+がコスパよく、逆にAWS ECSやGKEを既に運用しているチームはOSS版でCI/CD統合を自作したほうが柔軟です。GitHubのdagster-io/dagsterリポジトリにはHelmチャートとdocker-compose例が同梱されており、自前運用のスタート地点は用意されています。

本番運用のベストプラクティス

Dagsterを本番で3年ほど回してきた経験から、事故を避けるためのプラクティスをいくつか。どれも一度は痛い目にあったやつです。

  • Code Locationを分割する: 大規模プロジェクトではDefinitionsを機能ドメインごとに複数プロセスに分割します。1つのCode Locationが起動失敗しても他が生きるため、ブラストラディウスを小さくできます。
  • ジョブレベルではなくアセットレベルでリトライを設定: RetryPolicyを@assetに付けると、失敗したアセットだけを指数バックオフでリトライできます。ジョブ全体をリトライすると副作用のあるアセットが二重実行される事故が起きます。
  • センサーのrun_keyは必ず設定: 冪等性が破綻するとメトリクスが二重集計され、下流のダッシュボードが狂います。
  • アセットのFreshness Policyでデータ鮮度をSLA化: 「このアセットは3時間以内に更新されているべき」というSLAを宣言的に書けます。違反時にはSlackアラートを飛ばせます。
  • Panderaでのスキーマ検証を`@asset`関数の入口に挟む: Dagsterのアセットチェックとは別に、実行時の型を守る二重の防波堤になります。

FastAPIバックエンドの人間から見ると、Dagsterは「Pythonicな設計思想」と「データエンジニアリング固有の要求」の中間点に立っていて、思っていたよりずっと学びやすいツールでした。cron地獄から抜け出したい方は、まずDagster公式ブログの実例集を1本読んでみるといいと思います。

よくある質問

Dagsterは無料で使えますか？

DagsterのOSS版はApache 2.0ライセンスで完全に無料です。Webserver、Daemon、Kubernetes連携も含めてすべて自前でセルフホストできます。マネージド版のDagster+は従量課金で、実行時間や機能量に応じて課金されます。

DagsterとAirflowを併用できますか？

可能です。移行期にはdagster-airflowパッケージを使って、既存のAirflow DAGをDagsterジョブとしてラップできます。ただし本格運用ではどちらか一方に統一するほうが認知負荷が下がるため、段階的な移行計画を推奨します。

ソフトウェア定義アセットとは何ですか？

データパイプラインが生成する成果物（テーブル、モデル、ファイル、ダッシュボード）そのものをコード上のオブジェクトとして定義する仕組みです。Dagsterでは@assetデコレータを関数に付けることで、その関数の戻り値がアセットになります。

Dagsterの学習コストはどのくらいですか？

Pythonと型ヒントに慣れているエンジニアであれば、基本的なアセット定義とスケジュール設定は1日で書けるようになります。パーティション、IO Manager、Sensorの深い部分を使いこなすには1〜2週間の学習期間を想定しておくと安全です。

Dagsterはどのようなデータストアに対応していますか？

公式インテグレーションでは、Snowflake、BigQuery、Redshift、Postgres、DuckDB、S3、GCS、Azure Blob、Databricks、Sparkなど主要なストレージと計算エンジンに対応しています。dbt、Airbyte、Fivetran、Great Expectationsなどのモダンデータスタックとの統合も充実しています。

Tomás Oliveira
著者について Tomás Oliveira

Python backend developer who came to data work via FastAPI. Bridges the messy world between APIs and pipelines.

関連記事