okpy

Pythonエンジニア兼テックリーダーが、多くのプロジェクトとチーム運営から得た実践的な知識を共有するブログです。

🐍 Python dagster: 複雑なデータ処理、「手動」でまだ消耗していますか？

lib python

🐍 Python dagster: 複雑なデータ処理、「手動」でまだ消耗していますか？

📝 TL;DR (3行要約)

  • 何？: dagsterは、データ処理の流れ（パイプライン）を美しく定義し、実行し、監視するための次世代Pythonライブラリです。
  • いつ？: 複数のステップを持つデータ変換、ETL/ELT、機械学習のフィーチャー生成など、処理の順序と依存関係を厳密に管理したいときに使います。
  • 利点？: 処理の依存関係を「データ資産（Asset）」として明確化し、エラー発生時の追跡や再実行を劇的に簡単にしてくれます。

1. 🤔 一体dagsterとは何？（核心的な役割と主な使用例）

Pythonを使ったデータ処理は、データ分析、機械学習、Webサービス開発など、あらゆる分野で欠かせません。しかし、プロジェクトが大きくなるにつれて、「どのデータが、どの処理を経て、次の処理に使われたのか」を追跡するのが非常に難しくなります。

もしあなたが、以下のような悩みを抱えているなら、まさに今、dagsterがその解決策を提示してくれます。

  • 「昨夜実行したスクリプトが途中で止まったけど、どこまで進んだか分からない…」
  • 「Aというデータセットを作るには、必ずBとCの処理が終わっていないといけないのに、間違えてAだけ実行してしまった！」
  • 「処理の流れ全体を、誰が見ても分かるように図で示したいけど、毎回手書きでフローチャートを書いている…」

🏭 核心的な役割：データ処理の「工場長」と「レシピブック」

dagsterの核心的な役割は、データ処理を単なる「実行可能なスクリプト」としてではなく、「データ資産（Asset）」を生み出すプロセスとして捉え直すことです。

従来のデータ処理は、料理で例えるなら「単発のレシピ」でした。

  1. 野菜を切るスクリプトを実行。2. 肉を焼くスクリプトを実行。3. それらを組み合わせてカレーを作るスクリプトを実行。

もし途中で焦げ付いたら（エラーが出たら）、どこからやり直すべきか、どの材料（中間データ）がダメになったのか、自分で確認しなければなりません。

一方、dagsterは「高度に管理された工場長」のようなものです。

dagsterは、データ処理の各ステップを独立した「作業工程（Op）」として定義し、その工程が生み出す成果物（中間データや最終結果）を「データ資産（Asset）」として厳密に管理します。

比喩的に言えば、dagsterは単なるレシピではなく、「この肉は、このサプライヤーから仕入れ、この温度で焼き、その結果を次の工程に渡す」という、品質管理まで含めた詳細な「製造指示書」なのです。

これにより、以下の問題が解決します。

  1. 依存関係の明確化: 「Aを作るにはBとCが必須」という関係がコード内で明確に定義されます。dagsterはBとCが完了するまでAの実行を許可しません。2. 実行と監視: 処理がどこまで進んだか、エラーがどこで発生したか、どのデータに問題があるかを、専用のWeb UI（Dagit）でリアルタイムに監視できます。3. 再実行の効率化: エラーが発生しても、失敗したステップ以降の必要な部分だけを自動で特定し、効率的に再実行できます。

📊 主な使用例：このライブラリが真価を発揮する瞬間

dagsterは、データが複雑な流れを辿る、あらゆる状況で強力なツールとなります。

1. ETL/ELTパイプラインの構築

これは最も典型的な使用例です。 * E (Extract - 抽出): データベースやAPIからデータを取得する。 * T (Transform - 変換): 取得したデータをクレンジング、集計、結合する。 * L (Load - ロード): 変換後のデータをデータウェアハウス（例: Snowflake, BigQuery）に書き込む。

これらのステップが依存しあっているとき、dagsterは「抽出が完了しないと変換に進めない」「変換が終わらないとロードできない」という流れを保証し、各ステップの成功/失敗を記録します。

2. 機械学習モデルのトレーニングフロー管理

機械学習プロジェクトでは、単にモデルを学習させるだけでなく、その前後に多くのデータ処理が必要です。

  • データ準備: 生データから特徴量（Feature）を生成する。
  • トレーニング: 特徴量を使ってモデルを学習させる。
  • 評価: 学習済みモデルの性能を評価する。

特徴量の生成ステップが失敗すれば、トレーニングは無意味です。dagsterは、この特徴量やモデル自体を「Asset」として管理することで、「このモデルは、いつ生成された、どの特徴量セットで学習されたのか」を常に追跡可能にします。

3. 定期的なデータレポートの自動生成

毎週月曜日に、先週の売上データを集計し、グラフを作成し、Slackやメールで送信する、といった定型業務にも最適です。

集計処理、グラフ生成処理、通知処理をすべてdagsterのジョブとして定義しておけば、万が一集計処理でエラーが出ても、グラフ生成や通知が誤って実行されることを防ぎ、工場長（dagster）がすぐに「集計処理が失敗しました」と報告してくれます。

2. 💻 インストール方法

dagsterはPythonのパッケージ管理ツール pip を使って簡単にインストールできます。開発環境でWeb UI（Dagit）を使うために、dagster-webserver も一緒にインストールするのが一般的です。

以下のコマンドをターミナルで実行してください。

pip install dagster dagster-webserver

3. 🛠️ 実際に動作するサンプルコード

ここでは、最も基本的な「2つのデータ資産を生成し、それらを組み合わせて最終的なデータ資産を作る」というパイプラインを定義します。

このコードを repo.py として保存してください。

# repo.py

from dagster import asset, Definitions, job, op, define_asset_job

# --- 1. データ資産（Asset）の定義 ---

@asset(group_name="sources")
def raw_data():
    """
    ステップ1: 外部から生データを取得する処理をシミュレート。
    この処理が成功すると、リスト形式の「生データ」という資産が生成される。
    """
    print("➡️ 生データを取得中...")
    return [10, 20, 30, 40, 50]

@asset(group_name="preprocessing", deps=[raw_data])
def processed_data(raw_data):
    """
    ステップ2: 生データを受け取り、前処理（ここでは各要素を2倍）を行う。
    deps=[raw_data] により、raw_dataが先に実行されることが保証される。
    """
    print("🔄 前処理を実行中...")
    return [x * 2 for x in raw_data]

@asset(group_name="reporting", deps=[processed_data])
def final_report(processed_data):
    """
    ステップ3: 前処理済みデータを受け取り、最終レポート（合計値）を生成する。
    """
    print("✅ 最終レポートを作成中...")
    total = sum(processed_data)
    print(f"🎉 最終合計値: {total}")
    return total

# --- 2. 定義のまとめ ---

# dagsterに必要な情報（アセットの集合）をDefinitionsオブジェクトに渡す
defs = Definitions(
    assets=[raw_data, processed_data, final_report],
    # 定義したアセット全てを実行するジョブも自動で定義できる
    jobs=[
        define_asset_job(
            name="full_data_pipeline", 
            selection=["raw_data", "processed_data", "final_report"]
        )
    ]
)

実行方法（Web UIの起動）

このファイルがあるディレクトリで、以下のコマンドを実行してください。

dagster dev

ターミナルに表示されたURL（通常は http://127.0.0.1:3000）にアクセスすると、dagsterのWeb UIである Dagit が起動します。Dagit内で定義したジョブを選択し、「Launch Run」ボタンを押すことで、パイプラインが実行されます。

4. 🔍 コードの詳細説明

さて、上記のシンプルなコードですが、従来のPythonスクリプトとは全く異なる哲学に基づいています。dagsterのコードを理解する上で重要な3つの要素を、初心者の方にも分かりやすいように徹底的に解説します。

1. @asset デコレータとデータ資産の概念 (Asset)

従来のプログラミングでは、関数は「処理」そのものを表しました。しかし、dagsterでは @asset デコレータを関数に付与することで、その関数の役割を「特定のデータ資産を生み出す製造プロセス」に変えます。

@asset(group_name="sources")
def raw_data():
    # ... データ生成ロジック ...
    return [10, 20, 30, 40, 50]
  • 関数の名前 (raw_data): これは、生成されるデータ資産の名前になります。
  • 戻り値 (return): 関数の戻り値が、この資産の具体的な「値」となります。dagsterはデフォルトでこの値を永続化（保存）しようとします（ローカル実行では一時ファイルとして扱われます）。
  • group_name: UI上で資産を分類するためのラベルです。大規模なプロジェクトでは必須となります。

なぜこれが重要なのか？

データ資産として定義することで、dagsterは「どの資産が、どの資産に依存しているか」を自動で把握できます。これは、単なる関数呼び出し（processed_data(raw_data())）とは根本的に異なります。関数呼び出しの場合、実行順序は定義されますが、その結果（データ）が工場全体でどのように利用され、管理されているかは外部からは分かりません。

@asset を使うことで、dagsterはあなたのデータ処理全体を「データフローのグラフ」として内部的に構築し、そのグラフに基づいて実行計画を立てるのです。

2. 依存関係の定義 (deps と関数の引数)

dagsterにおける依存関係の定義方法は非常に直感的であり、Pythonの関数の仕組みを最大限に活用しています。

@asset(group_name="preprocessing", deps=[raw_data])
def processed_data(raw_data):
    # raw_dataという引数名で、前の資産の値を受け取る
    return [x * 2 for x in raw_data]
  • deps=[raw_data]: これは、この processed_data アセットが実行される前に、必ず raw_data アセットが成功裏に完了している必要があることを宣言しています。
  • 引数名 (raw_data): dagsterは賢く、引数名が依存するアセットの名前と一致していることを認識します。実行時、dagsterは先に生成された raw_data 資産の値を、この関数の引数として自動的に注入（Inject）してくれます。

この仕組みにより、開発者はデータの流れを心配することなく、目の前の「変換ロジック」だけに集中できます。dagsterがデータの受け渡しと依存関係の順守をすべて引き受けてくれるのです。

3. Definitionsとジョブの定義

定義した複数のアセットを、dagsterが認識し、実行可能な形にまとめるのが Definitions オブジェクトの役割です。

# dagsterに必要な情報（アセットの集合）をDefinitionsオブジェクトに渡す
defs = Definitions(
    assets=[raw_data, processed_data, final_report],
    jobs=[
        define_asset_job(
            name="full_data_pipeline", 
            selection=["raw_data", "processed_data", "final_report"]
        )
    ]
)
  • Definitions: これは、dagster環境（リポジトリ）のコンフィギュレーションファイルのようなものです。定義した全てのアセットをリスト形式で渡します。
  • define_asset_job: アセットは単体で存在しますが、それらを「実行可能な一連の処理」としてまとめるのが「ジョブ（Job）」です。ここでは、定義した3つのアセットをすべて含む full_data_pipeline という名前のジョブを作成しています。

この Definitions オブジェクトが、あなたが dagster dev コマンドを実行したときに、Dagit (Web UI) に表示される「設計図」そのものになります。

4. Dagit (Web UI) の役割の重要性

サンプルコードの実行方法で dagster dev を推奨したのは、dagsterの真価がこのWeb UI、Dagit にあるからです。

初心者が従来のスクリプト（例: python my_script.py）を実行するとき、結果はターミナルのログを見るしかありませんでした。しかし、Dagitは以下のような機能を提供します。

  • 可視化 (Asset Graph): 定義したアセット間の依存関係が、自動的に美しいグラフとして表示されます。どのデータがどこから来て、どこへ流れるのかが一目瞭然です。
  • 実行と監視 (Runs): ジョブの実行を開始し、その実行が成功したか失敗したか、失敗した場合はどのステップ（アセット）でエラーが発生したかを、リアルタイムで確認できます。
  • ログ管理: 各ステップの標準出力（print文など）がDagit内に整理されて保存されます。エラー発生時も、該当ステップのログにすぐにジャンプできます。

dagsterを使うということは、単にコードを書くことではなく、このDagitという「制御盤」を使ってデータ工場を管理することなのです。複雑な処理になればなるほど、Dagitの存在が開発効率を飛躍的に向上させます。

5. ⚠️ 注意点またはヒント

dagsterは強力ですが、従来のスクリプト実行とは異なるパラダイム（考え方）を要求します。初心者がスムーズに導入するために、特に重要なヒントを2つ紹介します。

1. データ永続化（I/O Manager）の概念を理解する

サンプルコードでは、raw_data アセットがリストを返し、processed_data アセットがそれを引数として受け取りました。ローカルでの実行では、このデータの受け渡しはメモリ内や一時ファイルで行われます。

しかし、実際のプロダクション環境では、データ処理は数時間かかることもあります。その際、中間データ（例: processed_data の結果）をメモリに保持し続けるのは現実的ではありませんし、もし途中でサーバーが落ちたらデータは消えてしまいます。

ヒント: dagsterは、アセットの戻り値を永続的に保存し、必要に応じてロードする仕組みとして「I/O Manager」を使います。

初心者のうちは意識しなくても動きますが、プロジェクトを本格化させる際には、dagster-pyspark や dagster-aws のような拡張機能を使って、S3やデータベースに中間データを保存するI/O Managerを設定する必要があります。

「中間データはメモリではなく、永続的なストレージに保存されるべきだ」という原則を常に意識してください。これが、従来の関数呼び出しとdagsterの大きな違いです。

2. アセットの「冪等性（べきとうせい）」を意識する

冪等性とは、「何度実行しても同じ結果が得られる」という性質を指します。

dagsterは、失敗したアセットだけを再実行したり、依存関係が変更されていないアセットの実行をスキップしたり（キャッシュのような振る舞い）することができます。これは、アセットが冪等性を持っているという前提のもとで最も効率的に機能します。

例えば、「現在時刻をログに書き込む」という処理は、実行するたびにログの内容が変わるため冪等ではありません。

一方、「特定の入力ファイルAを読み込み、常に同じ変換ロジックBを適用する」という処理は、入力ファイルAが変わらない限り、何度実行しても同じ出力が得られるため冪等です。

dagsterでアセットを定義する際は、「この関数は、入力が変わらない限り、何度実行しても同じ出力（データ資産）を生み出すか？」を自問自答し、冪等性を保つようにロジックを設計することが、パイプラインの安定性とデバッグの容易さにつながります。

6. 🔗 一緒に見ておくと良いライブラリ

dagsterと同じくデータパイプラインのオーケストレーションを担うライブラリとして、Python界で最も有名で歴史があるのが Apache Airflow です。

✈️ Apache Airflow

  • 役割: ジョブのスケジューリングとワークフロー管理。
  • dagsterとの違い: Airflowは「タスク（処理）」の実行順序を管理することに焦点を当てています。一方、dagsterは「データ資産（結果）」そのものに焦点を当て、データ中心の設計を促します。
  • 学習のヒント: Airflowは巨大なコミュニティと豊富なドキュメントがありますが、概念がタスク中心であるため、複雑なデータ依存関係を扱う際には、dagsterの「Asset」中心のアプローチの方が直感的だと感じる開発者が増えています。

もしあなたがデータオーケストレーションの歴史と主流な方法論を知りたいならAirflowを、もしあなたがモダンでデータ中心の設計哲学を学びたいならdagsterを、それぞれ深く掘り下げてみてください。両者の違いを理解することで、データエンジニアリングの全体像が見えてきます。

7. 🎉 まとめ

今日は、複雑なデータ処理の流れを美しく、効率的に管理するための強力なツール、dagsterをご紹介しました。

dagsterの核心は、単なるスクリプト実行ではなく、データ処理を「データ資産（Asset）」を生成・消費する工場として捉え直す点にあります。この新しいパラダイムによって、依存関係の管理、エラーの追跡、再実行が劇的に簡単になります。

あなたがもし、データ処理の複雑さにうんざりしているなら、今こそdagsterの「工場長」にあなたのデータパイプラインの管理を任せてみましょう。

🚀 次の挑戦課題

  1. Dagitを探検しよう: dagster dev を実行し、DagitのUIを隅々まで見てください。特に「Asset Graph」の表示が、あなたの書いたコードとどのように対応しているかを確認しましょう。2. エラーを意図的に発生させてみよう: raw_data 関数内で、意図的に raise Exception("データ取得失敗！") を追加し、ジョブを実行してみてください。Dagitがどのようにエラーをハイライトし、後続の processed_data や final_report の実行を防ぐか観察しましょう。

この一歩が、あなたのデータエンジニアリングスキルを次のレベルへと引き上げてくれるはずです。Happy Coding!

🔖 推奨タグ (ハッシュタグ)

  • データ分析

  • dagster