dagster入門｜Pythonのデータ処理を依存関係つきで可視化・部分実行する【競馬AI開発#58】

dagsterは、前処理や特徴量作成といった複数のデータ処理を「依存関係つきの一枚のグラフ」として可視化・実行できるPython製のデータオーケストレーションツールです。

この記事では、競馬予想AIの「データ前処理〜特徴量作成」を題材に、dagsterのasset（アセット）機能だけを使って、最小構成からdagsterを導入する方法を解説します。インストールからWeb UIの操作、実装コードまでを、実際の画面とあわせて一つずつ追っていきます。

‍■今回やること
前回の記事では、netkeiba.comからのデータ収集を「コマンド一発で実行できるスクリプト」に仕立て、運用しやすい形に整えました。

今回はその次の工程、つまり集めてきたデータを機械学習モデルが学習できる形に変える前処理〜特徴量作成の部分に「dagster」‍という便利なツールを導入します。

これまで前処理〜特徴量作成は、1枚のnotebookの中で、セルを上から順に実行することで回していました。dagsterを使うと、以下のように処理どうしの依存関係つきで、一枚の絵として可視化できます。

このグラフは手動で書いたものではありません。後ほど解説するように、dagsterが実装を読み取って自動でこのグラフを描いてくれます。

このUI上で処理を実際に実行することができ、さらにdagsterは処理どうしの依存関係を把握しているので、変更した箇所とその影響が及ぶ下流の処理だけを選んで実行することもできます。特徴量作成では、「ある特徴量の計算方法だけ少し変えて、結果を試したい」といった作り替えながらの実験を頻繁に行うため、そのような実験的な使い方にも便利です。

‍■競馬AIについて
netkeiba.comから収集したデータを機械学習モデルに学習させることで期待リターンを計算し、リターンが最大化するように自動投票を行う競馬予想AIを開発しています。

すでにシステムとしては一通り完成しており、現在はこの完成したシステムをベースに、特徴量の追加やモデルの改善など、さらなる精度向上に取り組んでいる段階です。

このシリーズでは、こうしたシステムをゼロから構築する過程を、コード付きで詳しく解説しています。ソースコードは、下に進んでダウンロード・解凍すれば使うことができるので、ぜひ実際にコードを動かしてみてください。

1. なぜdagsterを導入するのか

まずは、なぜ特徴量作成という工程に、dagsterが必要なのかを整理します。dagsterを使うと、主に次の3つができるようになります。

• 前処理から特徴量作成までの一連の処理を、依存関係つきの一枚のグラフとして可視化する。

• ボタンひとつで、すべての処理を正しい順番でまとめて実行する。

• 一部だけを作り替えたとき、変更した箇所とその下流だけを部分的に実行する。

特徴量作成は「全部を一気に実行したい」場面と「一部だけ作り替えて試したい」場面が混在する工程であり、dagsterはその両立を自然に解いてくれます。だからこそ、単純なスクリプト化ではなく、あえてdagsterを選ぶ価値があります。

ここ最近のこのシリーズでは、競馬AIのデータまわりの「運用のしやすさ」を少しずつ整えてきました。#56, #57の記事では、これまでnotebook上でセルを適宜修正しながら実行していたデータ収集処理をスクリプトにまとめ、ターミナルで1行コマンドを打つだけで完結できるようにすることで、再現性が高く、運用に組み込みやすい形にアップデートしました。

ところが、特徴量作成処理は単純にスクリプト化するだけでは、不都合が発生します。というのも、特徴量作成には‍「一部だけ作り替えて試す」という探索的・実験的な使い方が、依然として強く残るからです。
たとえば「ある前処理のロジックだけを少し変えて、特徴量への効き方を確かめたい」「新しい特徴量をひとつ足してみて、その部分だけ作り直したい」といった場面が日常的にあります。

しかし、スクリプトは基本的に‍「最初から最後まで全て実行される」‍という前提で動くので、一部分だけを変えて試したいときでも、毎回先頭から全部を実行し直すことになってしまます。

つまり、特徴量作成では、

• ‍定型処理として全部を一気に実行したい場面と、

• ‍実験として一部だけを作り替えて実行したい場面

の両方を、無理なく両立させたいわけです。

この両立を自然に解いてくれるのが、今回導入するdagsterの‍asset（アセット）‍という仕組みです。これを使うと、‍処理どうしの依存関係をツール側に管理させることで、「全部を依存順に実行する」ことも、「変更した部分とその影響範囲だけを実行する」ことも、どちらも同じ仕組みの上でできるようになります。

1.1. 「データオーケストレーションツール」とは

dagsterは、ひとことで言えばデータオーケストレーションツールと呼ばれる種類のソフトウェアです。

データ分析や機械学習の開発では、複数の処理を決まった順番で実行する必要があります。最初のうちは手作業で順番にスクリプトを叩けば十分ですが、処理の数が増えてくると、だんだん次のような問題が出てきます。

• どの処理がどの処理の後でなければならないか（依存関係）を、人間が覚えておく必要がある

• どこかで失敗したとき、どこから再実行すればいいか分かりにくい

• 「今あるデータは最新なのか、古いのか」が分からなくなる

データオーケストレーションツールは、こうした「処理の段取りと状態の管理」をまるごと引き受けてくれます。

この分野で最も有名なのがApache Airflowですが、dagsterはそれより新しく、設計思想が違います。Airflowが「タスク（処理の手順）」を並べてパイプラインを組み立てるのに対し、dagsterは「asset（生成されるデータ）」を起点にパイプラインを定義します。また、ローカルマシンで気軽に動かして試せる開発体験の良さも特徴です。

dagsterとAirflowの違いを表に整理すると、次のとおりです。

項目         | dagster                      | Airflow                        
-------------+------------------------------+--------------------------------
中心概念     | asset（生成されるデータ）    | task（実行する処理）           
組み立て方   | データの依存関係でつなぐ     | 処理の手順でつなぐ             
変更時の実行 | 変えたassetと、その下流だけ  | 基本は手順を最初から実行       
依存の可視化 | 実装から依存グラフを自動生成 | DAGを記述して定義              
ローカル開発 | dagster dev で手軽に起動     | 相対的にセットアップが重め     
登場時期     | 2019年ごろ〜（後発/DX重視）  | 2014年ごろ〜（実績/周辺が豊富）

Airflowが「処理（タスク）」を主役にするのに対し、dagsterは「データ（asset）」を主役にする——この発想の違いが、次に説明するasset志向の核心です。なお、assetをはじめとするdagsterの公式な概念解説は、dagster公式ドキュメント（Defining assets）にまとまっています。本記事はそのエッセンスを、競馬AIの特徴量作成という実例に沿って解説していきます。

1.2. asset志向: 主役は「手順」ではなく「データ」

dagsterを理解するうえで最も大事なのが、このasset（アセット）という考え方です。

主役を「手順」ではなく「生成されるデータ」に置くのです。

• population（学習母集団のテーブル）

• results（前処理済みのレース結果テーブル）

• features（最終的な特徴量テーブル）

といった、出来上がるデータのひとつひとつがassetです。「この処理を実行する」ではなく「このデータを作る」という宣言の集まりとして、パイプライン全体を組み立てます。

一見すると小さな違いに思えますが、これが効いてきます。「データ」を主役にすると、dagsterは「このデータはどの処理から生まれ、どのデータに使われるのか」という系譜（lineage、リネージ）を自動的に把握できます。だからこそ、冒頭でお見せしたような依存グラフが自動で描かれ、「どこを変えると、どこに影響するか」がひと目で分かるのです。

1.3. dagsterのメリット: ：依存関係を知っているから「部分実行」できる

assetを定義するとき、私たちは「このassetは、どのassetを必要とするか」という依存関係も一緒に宣言します（具体的な書き方は後で解説）。すると、dagsterは全assetの依存関係をグラフ（DAG＝有向非巡回グラフ）として保持します。この「依存関係を知っている」ことが、各使用ケースでの実行に役立ちます。

まず「‍定型処理として全部を一気に実行したい場面」では、ボタンひとつで全assetを順番に実行（materialize）できます。

ポイントは二つ目の「‍実験として一部だけを作り替えて実行したい場面」においてです。あるassetを作り替えたとき、dagsterは「そのassetに依存している下流のassetはどれか」を知っているので、変更したassetとその下流だけをピンポイントで実行できます。

さらにdagsterは、上流のassetを更新すると、それに依存する下流のassetを「Unsynced（同期されていない＝古い状態）」として表示してくれます。

「どれを作り直すべきか」をツールの側が教えてくれるので、全部を闇雲にやり直す必要がなくなります。これが、特徴量作成のような「一部だけ作り替えて何度も試す」工程と、とても相性が良いわけです。

2. dagsterを動かす：導入とWeb UIの操作

ここからは、実際に手元でdagsterを動かす手順と、いま使えるUIの機能を、画面を見ながら一つずつ紹介していきます。

2.1. dagsterを導入し、UIを起動する

本プロジェクトはパッケージ管理に uv を使っている（uvの導入・使い方は#55で詳しく解説しています）ので、dagsterの導入は次のコマンドで行います。

uv add dagster dagster-webserver

• dagster … asset定義やCLIなど、dagster本体。

• dagster-webserver … ブラウザで見るWeb UIを提供するサーバー。

インストールできたら、UIを起動します。本プロジェクトでは Makefile に起動コマンドをまとめてあるので、v3_6_0/src/ ディレクトリで次のコマンドを実行するだけです。

make dagster

これは内部で、次のコマンドを実行しています。

uv run dagster dev -m pipeline.definitions

dagster dev は、開発用にdagsterの一式をまとめて立ち上げるコマンドです。‍-m pipeline.definitions は「pipeline/definitions.py に書いた定義を読み込んでね」という指定です（この中身は次の章で解説します）。

起動すると、ターミナルに次のようなメッセージが表示されます。

Serving dagster-webserver on http://127.0.0.1:3000

あとは http://localhost:3000 をブラウザで開けば、冒頭で見たグラフが表示されます。

2.2. Asset Lineage: 全体像を把握

最初に見てほしいのが、冒頭でもお見せした Lineage（アセットの依存グラフ）の画面です。左のタブの「ineage」を開くと表示されます。

コードを読まなくても、「どの処理が何に依存しているか」「最終的に何ができあがるか」が、ひと目で把握できます。これがasset化の最初のメリットです。

2.3. Materialize all: 全部まとめて実行する

次に、実際に処理を動かしてみます。画面右上の Materialize all ボタンを押すと、グラフ上の全assetが、依存関係の順番どおりに実行（materialize）されます。01_create_features.ipynbの処理を上から下まで流したのと、同じことが起きます。

左の「Runs」タブを開くと、処理の進み具合がより詳細にリアルタイムで表示されます。

どのassetが今動いているのか、それぞれにどれくらい時間がかかったのか、ログには何が出ているのかが、すべて追えます。

すべて終わると、各assetが成功した状態で表示されます。

2.4. 個別assetを部分実行する

グラフ上のassetをクリックすると、そのassetの詳細を見られます。

「このassetは何をするものか」がUI上で確認できるわけです。あわせて、依存しているasset、過去にいつ実行されたか、といった情報も並びます。

そして、この画面から「Linage」を選択すると、

• Upstream: そのアセットと上流のアセット全て

• Downstream: そのアセットと下流のアセット全て

• Nearest Neighbors: そのアセットと、一つ近隣の上流・下流アセット

を選んで部分的に実行することもできます。（これがめちゃくちゃ便利！）

たとえば「horse_results の作り方を変えた」というとき、horse_results とその下流の features だけを選んで実行すれば済みます。無関係な population や results を、もう一度走らせる必要はありません。

そして、この「どこを実行すべきか」の判断を助けてくれるのが、 Unsynced の表示です。上流のassetを再実行すると、それに依存する下流のassetに「Unsynced」というマークが付き、「上流が変わったので、ここはもう古いですよ」と教えてくれます。

このマークを目印に、古くなったところだけを選んで作り直せばよいので、全部を最初からやり直す必要がありません。

3. dagsterの実装：assetの書き方と依存関係（deps）の宣言

では、上記のような「dagsterで実行できる仕組み」を作るには、どんな実装が必要なのでしょうか？結論から言うと、かなりシンプルなコードで、既存の処理をほとんど書き換えずに実現することができます。

3.1. ディレクトリ構成

今回のdagster関連のコードは、‍v3_6_0/src/pipeline/‍ という新しいディレクトリにまとめています。全体の構成は次のとおりです。

v3_6_0/src/
├── pipeline/                  ← 今回追加したdagster関連ディレクトリ
│   ├── __init__.py            （空ファイル。pipelineをパッケージとして認識させる）
│   ├── assets.py              （assetの定義。今回の主役）
│   └── definitions.py         （assetをdagsterに登録する）
├── modules/                   ← 既存の処理本体（変更なし）
│   ├── preprocessing.py       （各種の前処理関数）
│   └── train/
│       ├── create_population.py    （母集団の作成）
│       └── feature_engineering.py  （特徴量作成 FeatureCreator）
├── Makefile                   ← dagster dev を起動するコマンド
└── 01_create_features.ipynb   ← 元になったnotebook

今回新しく足したのはpipeline/ と Makefileのみで、そのほかは元のコードそのままです。

3.2. assets.py: アセット定義の本体

今回の主役が、このassets.pyです。ここにアセットを登録することで、前章で見たような仕組みが8割方完成します。

‍▼v3_6_0/src/pipeline/assets.py

import dagster as dg
from modules import preprocessing
from modules.train import create_population
from modules.train.feature_engineering import FeatureCreator


@dg.asset
def population() -> None:
    """学習・検証・テスト母集団 (race_id, date, horse_id) を作成する。"""
    create_population.create(from_="2020-01-01", to_="2025-12-31")


@dg.asset(deps=[population])
def results() -> None:
    """レース結果テーブルの前処理。"""
    preprocessing.process_results()


@dg.asset(deps=[population])
def race_info() -> None:
    """レース情報テーブルの前処理。"""
    preprocessing.process_race_info()


@dg.asset(deps=[population])
def horse_results() -> None:
    """馬の過去成績テーブルの前処理。"""
    preprocessing.process_horse_results()


@dg.asset(deps=[population])
def peds() -> None:
    """血統テーブルの前処理。"""
    preprocessing.process_peds()


@dg.asset
def jockey_leading() -> None:
    """騎手リーディングテーブルの前処理。"""
    preprocessing.process_jockey_leading()


@dg.asset
def trainer_leading() -> None:
    """調教師リーディングテーブルの前処理。"""
    preprocessing.process_trainer_leading()


@dg.asset
def sire_leading() -> None:
    """種牡馬リーディングテーブルの前処理。"""
    preprocessing.process_sire_leading()


@dg.asset(deps=[population])
def return_tables() -> None:
    """払い戻しテーブルの前処理。回収率シミュレーションで使う。"""
    preprocessing.process_return_tables()


@dg.asset(
    deps=[
        population,
        results,
        race_info,
        horse_results,
        jockey_leading,
        trainer_leading,
        peds,
        sire_leading,
    ],
)
def features() -> None:
    """特徴量作成処理を実行し、populationテーブルに全ての特徴量を結合する。"""
    FeatureCreator().create_features()

やっていることは驚くほどシンプルで、元々notebook上で実行していた処理を@dg.asset というデコレータで包んでいるだけです。

たとえば一番上のpopulationを見てみます。

@dg.asset
def population() -> None:
    """学習・検証・テスト母集団 (race_id, date, horse_id) を作成する。"""
    create_population.create(from_="2020-01-01", to_="2025-12-31")

中身の create_population.create(...) は、これまでnotebookのセルで呼んでいたのと同じ、母集団を作る関数です。その関数呼び出しの上に ‍@dg.asset を付けるだけで、この関数が「populationというassetを作る処理」としてdagsterに認識されます。関数名（population）が、そのままassetの名前になります。

次に、依存関係の宣言（deps）を見ます。

@dg.asset(deps=[population])
def results() -> None:
    """レース結果テーブルの前処理。"""
    preprocessing.process_results()

results（レース結果の前処理）には、deps として population が指定されています。これは「resultsを作る前に、populationが必要ですよ」という依存関係の宣言です。この指定をするだけで、dagsterは「population → results」という矢印をグラフに描いてくれます。コードに書いた依存関係が、そのままUIのグラフになるわけです。

3.3. definitions.py: dagsterへの登録

assetを定義しただけでは、dagsterはまだそれを見つけられません。「これらのassetが、私たちのパイプラインですよ」と登録する必要があります。それを行うのがdefinitions.pyです。

‍▼v3_6_0/src/pipeline/definitions.py

"""
dagster dev のロード対象となる Definitions を公開する。
"""

import dagster as dg

from . import assets

defs = dg.Definitions(assets=dg.load_assets_from_modules([assets]))

load_assets_from_modules([assets]) が、assets.pyの中に定義された全assetを自動でかき集めてくれる関数です。これによって、dagsterが認識できる形になります。

最後に、起動を楽にするためのMakefile（＝コマンドを簡単に呼び出せるように登録しておくもの）を作成しておきます。

‍▼v3_6_0/src/Makefile

# dagster dev 起動
export DAGSTER_HOME := $(CURDIR)/.dagster_home

.PHONY: dagster

dagster: | $(DAGSTER_HOME)
	uv run dagster dev -m pipeline.definitions

$(DAGSTER_HOME):
	mkdir -p $@

これで、make dagster と打つだけで dagster dev が起動するようにしてあります。ここでひとつ大事なのが、DAGSTER_HOME という環境変数を固定している点です。

これを設定しないと、dagsterは一時ディレクトリを使うため、dagster dev を終了するたびに実行履歴が消えてしまいます。ここでは src 直下の .dagster_home フォルダに固定することで、「いつ・どのassetを実行したか」という履歴が、次回起動後も残るようにしています。保存場所が毎回ブレないので、状態の管理がしやすくなるわけです。

これで、ターミナルから以下のように実行すると、先ほどのようにdagsterを利用することができます。

cd v3_6_0/src/
make dagster

4. まとめと今後の展望

今回は、これまでnotebookで行っていた前処理〜特徴量作成の工程に、データオーケストレーションツールのdagsterを導入しました。
特徴量作成のように、定型的に全実行したい場面と、実験的に一部だけ作り替えたい場面が混在する工程では、こうしたasset化はとても有用です。

そして実はdagsterには、この土台の上に積み上げられる便利な機能が、まだまだたくさん用意されています。代表的なものでいうと、

• Partitions: assetを「年ごと」「日付ごと」等に分割しておく機能。例えば「直近半年分の新規データを学習に追加したい」といった場面に有用そう。

• Run Config: パラメータを、コードを書き換えずにUIのフォームから指定して実行できる。今回作成したasset.pyは学習期間を固定しているが、これを使えばUI上から指定できる。

• Asset Checks: assetを作ったあとに「欠損がないか」「重複レースが混ざっていないか」等を自動で検証する仕組み。

• メタデータの記録: materializeのたびに「使ったデータ件数」「モデルの精度スコア」「作った特徴量の数」等を自動で記録し、UI上で実行ごとの推移をグラフで追える機能。

• Schedules: 定期実行ができる機能。

• Sensors: 「新しいレースデータが届いたら自動で処理を起動」といったイベントベースの実行ができる機能。

• I/O Managers: 各assetが自分でファイルを読み書きしてくれる仕組み。現状、ローカルにcsvを保存先としているが、dagster上で保存することもできる。

といった機能が用意されています。これらを少しずつ組み合わせていくと、最終的には、データ収集 → 前処理 → 特徴量作成 → 学習 → 評価 → 予測 → 自動投票という競馬AIの全工程が、ひとつの依存グラフにつながり、UI上で全て運用できる未来が現実的に見えてきます。今回作ったのは、その構想の中の最初のひとかたまりというわけです。

5. ソースコードのダウンロード

完成した競馬AIのソースコードは、以下の記事に最新版が添付されているので、そこからダウンロードすることができます。

上記記事には、今回の記事で紹介したソースコード含め、データ収集・機械学習モデル構築・サーバー上での自動投票など、本格的な競馬AIの運用に必要な全工程のコードが含まれています。

また、この【競馬AI開発】シリーズでは、#1から上記コードの作り方を詳しく解説しているので、ご自身で競馬予想AIを開発・活用してみたいとお考えの方は、ぜひ過去の記事も併せてご覧ください。

dagster入門FAQ

最後に、ここまでの内容の復習も兼ねて、よくある質問（Q&A）のかたちで整理しておきます。

Q. dagsterとは？（どんなツール？）‍
A. dagsterは、データの前処理やモデル学習といった複数の処理を、依存関係つきのパイプラインとして定義・実行・監視できるPython製のデータオーケストレーションツールです。「実行する処理（タスク）」ではなく「生成されるデータ（asset）」を起点に組み立てるのが最大の特徴です。

Q. dagsterとAirflowの違いは？‍
A. 最大の違いは「主役」です。Airflowはtask（実行する処理）を並べてパイプラインを組み立てるのに対し、dagsterはasset（生成されるデータ）を起点に組み立てます。そのためdagsterは、変更したデータとその下流だけを賢く再実行でき、依存関係も実装から自動でグラフ化されます。Prefectはより手軽なPythonワークフロー寄りで、3つのなかではdagsterが最もデータ資産（lineage）を重視した設計です。

Q. dagsterのmaterialize（マテリアライズ）とは？‍
A. materializeとは、assetに対応する処理を実際に実行して、データを生成・更新することです。assetの定義が「設計図」だとすれば、materializeは「その設計図どおりに実際にモノを作ること」にあたります。

Q. dagsterのassetとtask（処理）の違いは？‍
A. assetは「作りたいデータ（テーブルやモデルなど）」そのものを宣言する単位で、taskは「実行する手順」を並べる単位です。dagsterではassetを宣言すると、その生成処理と依存関係がセットで管理され、UI上に依存グラフとして現れます。

Q. dagsterの導入・インストール方法は？‍
A. uvを使う場合は、dagster本体とWeb UI用のdagster-webserverを依存に追加し、dagster dev コマンドで起動します。あとはブラウザで localhost:3000 を開けば、依存グラフのUIが表示されます。具体的な手順は本記事の2章で解説しています。

Q. dagsterは無料で使える？‍
A. はい、dagster本体はオープンソースで無料で使えます（2026年6月時点）。チームでの運用やホスティングを支援する有償版のDagster+も提供されていますが、本記事のようにローカルで動かす範囲ではOSS版だけで完結します。

Q. 個人開発・小規模でもdagsterを使う意味は？‍
A. あります。とくに本記事の特徴量作成のように、「全部をまとめて実行したい」ときと「一部だけ作り替えて試したい」ときが混在する工程では、部分実行と依存関係の可視化が効いてきます。逆に処理が数個程度で今後も増えないなら、無理に導入する必要はありません。