Is Dagster free? Self-hosting Dagster vs Dagster+
Yes, Dagster is free. The open-source core is Apache 2.0, and you can run all of it on your own servers without paying Dagster Labs anything. That answers the licence question. It doesn't answer the budget question.
The budget question is whether you'd rather pay Dagster Labs or pay with your own hours. I self-host a production Dagster instance on a homelab Docker Swarm. It runs 112 assets, 57 schedules and 3 sensors for one person, me. Here is what the paid product costs today, what self-hosting has cost me in time, and a minimal setup you can start from.
Is Dagster free? What open source gives you
The open-source packages give you the whole orchestrator: assets, partitions, schedules, sensors, asset checks, the run queue and the web UI. Nothing in my pipeline needed a paid feature. The Dagster repository is licensed Apache-2.0.
What you take on is running it. A self-hosted Dagster has four parts:
1. Postgres for run history, event logs and schedule state. SQLite works on a laptop; I use it for local development only.
2. Code locations: containers that load your definitions and serve them over gRPC.
3. dagster-webserver: the UI and the GraphQL API.
4. dagster-daemon: schedules, sensors, the run queue, retries and run monitoring.
None of these is hard to start. Keeping them healthy for a year is the real work.
What Dagster+ costs
I checked dagster.io on 1 October 2026:
| Plan | Price | Includes | Per credit |
|---|---|---|---|
| Solo | $10 / month | 1 user, 1 code location, 1 deployment | $0.040 |
| Starter | $100 / month | up to 3 users, 5 code locations, 1 deployment | $0.035 |
| Pro | contact sales | unlimited seats, code locations and deployments | custom |
Serverless compute was $0.010 a minute on Solo and Starter. Solo and Starter had a 30-day free trial. The page defines a credit as one asset materialization or one op execution, and says a Hybrid deployment, where your own infrastructure runs the code, has no compute charge. There is no free tier on the page, only the trial.
To see what credits mean for you, count materializations. My pipeline has 112 assets, 36 of them partitioned (34 daily). Every daily partition of every daily asset is a credit, every day, plus every retry. Do that sum for your own graph before you assume a plan fits. I haven't priced my pipeline on Dagster+, so I won't guess my number.
---
This is the first part. The full post — including the rest of the working details — is on my site: Is Dagster free? Self-hosting Dagster vs Dagster+
図で理解する Dagster on ECS の構成と、失敗に気づく仕組み
セーフィー株式会社のデータ分析基盤グループでエンジニアをしている渡部です。
セーフィーのデータ分析基盤では、データソースごとにばらばらだったデータ取り込み処理のやり方を揃えるため、Python ライブラリの dlt(data load tool)を採用することとなりました。
dlt は取り込みを実行するだけなので「いつ実行するか？」「失敗にどう気づくか？」の仕組みは別で用意する必要があります。
そこで、オーケストレーションツールとして、Dagster も合わせて導入することとなりました。
現在は、プロダクトの DB から Snowflake へデータを取り込む処理の一部を ECS 上の Dagster で動作させています。
この記事では、そんな Dagster を理解するために、ECS の構成と失敗に気づく仕組みを図を使って説明することを目指します。
!
次のことはこの記事では扱いません。
アセット・スケジュール・センサーといった最も基本的な Dagster のコンセプト
アセットを定義する Python コードの書き方(dagster.Component など)
dlt の使い方
!
内容は dagster 1.13.13、dagster-aws / dagster-dlt 0.29.13、dlt 1.29.0 の時点のものです。
ECS の構成
まずはインフラ全体の構成です。
図の Mermaid コード(再生成用)---
config:
flowchart:
wrappingWidth: 400
---
flowchart LR
user@{ icon: "mdi:web", form: "square", label: "利用者のブラウザ", pos: "b", h: 48 }
subgraph aws["AWS アカウント(データ分析基盤)"]
waf@{ icon: "aws:waf", form: "square", label: "WAF", pos: "b", h: 48 }
alb@{ icon: "aws:elb", form: "square", label: "ALB<br/>(Cognito で認証)", pos: "b", h: 48 }
subgraph vpc["VPC"]
subgraph ecs["ECS on Fargate"]
web@{ icon: "aws:fargate", form: "square", label: "webserver<br/>(常駐)", pos: "b", h: 48 }
daemon@{ icon: "aws:fargate", form: "square", label: "daemon<br/>(常駐)", pos: "b", h: 48 }
cl@{ icon: "aws:fargate", form: "square", label: "code-location<br/>(常駐)", pos: "b", h: 48 }
rw@{ icon: "aws:fargate", form: "square", label: "run worker<br/>(run ごとに起動)", pos: "b", h: 48 }
end
db@{ icon: "aws:aurora", form: "square", label: "Aurora PostgreSQL<br/>(Dagster DB)", pos: "b", h: 48 }
re@{ icon: "aws:lattice", form: "square", label: "Resource Endpoint<br/>(ソース DB ごと)", pos: "b", h: 48 }
nat@{ icon: "aws:nat", form: "square", label: "NAT Gateway", pos: "b", h: 48 }
end
s3@{ icon: "aws:s3", form: "square", label: "S3<br/>(compute logs)", pos: "b", h: 48 }
end
subgraph src["ソース側の AWS アカウント"]
srcdb@{ icon: "aws:aurora", form: "square", label: "Aurora<br/>(ソース DB)", pos: "b", h: 48 }
end
subgraph saas["SaaS"]
sf@{ icon: "logos:snowflake-icon", form: "square", label: "Snowflake", pos: "b", h: 48 }
dbt@{ icon: "logos:dbt-icon", form: "square", label: "dbt Cloud", pos: "b", h: 48 }
slack@{ icon: "logos:slack-icon", form: "square", label: "Slack", pos: "b", h: 48 }
end
user --> waf --> alb --> web
web -- gRPC --> cl
daemon -- gRPC --> cl
daemon -- RunTask --> rw
ecs -- 5432 --> db
ecs --> s3
rw --> re -- VPC Lattice --> srcdb
ecs --> nat
nat --> sf
nat --> dbt
nat --> slack
style aws fill:#ffffff,stroke:#E7157B,stroke-width:2px
style src fill:#ffffff,stroke:#E7157B,stroke-width:2px
style vpc fill:#ffffff,stroke:#8C4FFF,stroke-width:2px
style ecs fill:#FFF8F0,stroke:#ED7100,stroke-width:1px,stroke-dasharray:5 5
style saas fill:#F7F7F7,stroke:#7D8998,stroke-width:1px
ECS の構成は、公式ドキュメントにある構成例(Deploying Dagster to Amazon Web Services)に沿っていて、とってもシンプルです。
ECS の上で動いているものは以下の 4 種類となっています。
ECS サービス / ECS タスク
役割
起動コマンド
code-location
アセットやスケジュールを定義したコードを読み込み、webserver と daemon の問い合わせに答える
dagster code-server start -h 0.0.0.0 -p 4000 -m data_integration.definitions
daemon
スケジュールやセンサーを評価して、run を起動する
dagster-daemon run -w workspace.yaml
webserver
UI を提供する
dagster-webserver -h 0.0.0.0 -p 3000 -w workspace.yaml
run worker
データ取り込みの処理 = run を 1 つ実行する
dagster api execute_run <JSON>(EcsRunLauncher という Python クラスが面倒を見てくれる)
このうち code-location・daemon・webserver は常に起動している(常駐している)ECS タスクです。
一方で run(ジョブの 1 回の実行)を受け持つ run worker はアドホックに立ち上がる ECS タスクです。
現在の dagster.yaml の設定では、Dagster DB に実行履歴などが保存されており、標準出力・標準エラー出力は S3 にファイルとしてアップロードされるようにしています。
利用者はロードバランサー経由で webserver にアクセスすれば、ブラウザでこれらを確認したり、アセットを手動で実行することができます。
取り込みを実行する run worker は、dlt を使って、以下のような流れでデータソースである Aurora PostgreSQL からデータを Snowflake に連携しています。
VPC Lattice の Resource Endpoint 経由で別 AWS アカウントにある Aurora PostgreSQL に接続してデータを取得
NAT Gateway から Snowflake にデータを書き込み
取り込みが終わったあとの変換は dbt Cloud のジョブに任せていて、その起動も Dagster から行っています。
Dagster で処理が実行されるまでに起きること
Dagster で処理が実行されるまでは、次の図の① → ③の流れになっています。
図の Mermaid コード(再生成用)---
config:
flowchart:
wrappingWidth: 400
---
flowchart LR
subgraph web["webserver(常駐)"]
ui["UI / GraphQL"]
end
subgraph daemon["daemon(常駐)"]
sch["スケジュールとセンサーを評価"]
launcher["EcsRunLauncher"]
end
subgraph cl["code-location(常駐)"]
defs["Definitions<br/>(アセット・ジョブ・スケジュール・センサー)"]
end
subgraph rw["run worker(run ごとに起動)"]
step["アセットの実行"]
end
pg[("Dagster DB")]
ui -- "① gRPC で定義を取得" --> defs
sch -- "① gRPC で定義を取得" --> defs
ui -- "② run をキューに積む" --> pg
sch -- "② run をキューに積む" --> pg
pg -- "キューから取り出す" --> launcher
launcher -- "③ ECS RunTask で起動" --> step
step -- "イベントを書き込み" --> pg
Dagster では、処理の定義(アセットやジョブ = Python のコード)があるのに対して、ジョブの 1 回の実行を表すのが run です。
ここら辺の用語を正しく使わないと理解が難しくなるため、少し整理しておきます。
用語
定義か実行か
何か
アセット
定義
作りたいデータ(テーブルなど)と、その作り方
ジョブ
定義
一緒に実行するアセットの組み合わせ
run
実行
ジョブの 1 回の実行。状態とログを持つ
step
実行
run の中の、アセット 1 つ分の実行
run が作られてから実行されるまでは、次の 3 段階(作る・積む・起動する)があることがわかります。
① run を作る
webserver か daemon が code-location に問い合わせて、run を作ります。
このとき、実行計画(どの step をどの順で実行するか)と Docker イメージが run に記録されます
② run をキューに積む
作った run を、Dagster DB に QUEUED の状態で書き込みます。
ここまでは UI から実行したときは webserver、スケジュールやセンサーのときは daemon が担当します。
③ run を起動する
起動自体は daemon が行います。
キューから run を取り出して、run に記録されたイメージで run worker の ECS タスクを作ります(EcsRunLauncher)
Dagster の挙動を把握するコツ
Dagster はワークフローエンジンということもあって、解説記事は多くありません。しかも、いろいろな環境にデプロイできて、処理も Python で書けることであればなんでもできてしまいます。インフラ構成・設定・実装までを考えると自由度が高くとても複雑に見えます。
全体を一度に見てしまうと複雑に見えてしまうので 1 つずつ分解して見ることが大切です。
一番メインとなるのは ECS サービス / ECS タスクです。ECS の構成のテーブルに記載した、起動コマンドに目を向けてみてください。
webserver と daemon は単に Dagster で用意されているコマンドを実行しているだけであり、設定ファイルである dagster.yaml と workspace.yaml だけを考えれば良いです。
一方で自分たちの Python コードが関連してくるのは code-location と run worker だけなのです。
エラー・失敗に気づく仕組み
ここまでで実行されるまでの流れを見てきました。では、失敗した時に気づけるようにするためにはどうすれば良いでしょうか？
現在、成功時・失敗時ともに社内の Slack チャンネルに通知を行なっていますが、ここからはどのように失敗に気づけるようにしたかにフォーカスしていきたいと思います。
さまざまなデータソースと連携できる Dagster では多種多様な失敗が起き得ます。
私が実際に遭遇したもので思い返せるものでも、以下のようなものが挙げられます。
Python が例外を出してエラーになるもの
メモリ不足で OOM キラーに叩き落とされるもの
エラーにならずにハングしてずっと処理が続くもの
daemon や code-location など、Dagster そのものが落ちるもの
Python の例外ならば、Dagster 内部からの監視・通知でカバーできますが、Dagster それ自体が落ちてしまうと失敗に気づくことすらできません。
そこで、次の図のように監視・通知の経路は 2 本作成してあります。
図の Mermaid コード(再生成用)---
config:
flowchart:
wrappingWidth: 400
---
flowchart LR
subgraph dagster["1 本目の経路: Dagster が通知する(daemon と code-location が動いている間だけ)"]
daemon@{ icon: "aws:fargate", form: "square", label: "daemon", pos: "b", h: 48 }
cl@{ icon: "aws:fargate", form: "square", label: "code-location<br/>(run_failure_sensor)", pos: "b", h: 48 }
daemon -- 定期的に呼び出し --> cl
end
subgraph aws["2 本目の経路: AWS が通知する(Dagster が落ちていても通知される)"]
ecsfail@{ icon: "aws:ecs", form: "square", label: "ECS タスクの<br/>異常終了・起動の失敗", pos: "b", h: 48 }
eb@{ icon: "aws:eventbridge", form: "square", label: "EventBridge", pos: "b", h: 48 }
zero@{ icon: "aws:ecs", form: "square", label: "常駐の ECS タスクが<br/>0 台のまま", pos: "b", h: 48 }
cw@{ icon: "aws:cloudwatch", form: "square", label: "CloudWatch アラーム", pos: "b", h: 48 }
sns@{ icon: "aws:sns", form: "square", label: "SNS", pos: "b", h: 48 }
q@{ icon: "aws:q", form: "square", label: "Amazon Q Developer", pos: "b", h: 48 }
ecsfail --> eb --> sns
zero --> cw --> sns
sns --> q
end
slack@{ icon: "logos:slack-icon", form: "square", label: "Slack", pos: "b", h: 48 }
cl -- Incoming Webhook --> slack
q --> slack
style dagster fill:#ffffff,stroke:#7D8998,stroke-width:1px
style aws fill:#ffffff,stroke:#E7157B,stroke-width:2px
1 本目の経路は、Dagster 内部からのものです。run_failure_sensor センサーを使用しています。
センサーの定期実行を支えているのは daemon、センサーの本体の判定ロジックの Python が動くのは code-location です。
内部からの監視・通知なので、daemon か code-location が落ちていると、失敗の通知が届きません。
2 本目の経路では、Dagster の外からの通知を用意しました。こちらは AWS の機能を使っています。
EventBridge: ECS タスクの異常終了と起動の失敗を検知する
CloudWatch アラーム: 常駐サービスの ECS タスクが 0 台になったことを検知する
どちらの通知も、SNS から Amazon Q Developer(旧 AWS Chatbot)を経由して Slack に通知されます。
以上、どの失敗をどちらの経路で拾うかをまとめると、次のようになります。
失敗
気づく経路
Python の例外
1 本目
メモリ不足(OOM)
2 本目(EventBridge)と 1 本目の両方
ハング
1 本目(30 分で FAILURE にする)
Dagster そのものが落ちる
2 本目(EventBridge と CloudWatch アラーム)
運用して分かった注意点
ここまでが失敗に気づく仕組みです。ここからは、運用してみて分かった注意点を 3 つほど書かせてもらいます。
どの注意点も run の状態の変わり方が関わるので、先に図にしておきます。
図の Mermaid コード(再生成用)---
config:
themeCSS: "foreignObject { overflow: visible; } foreignObject div { max-width: none !important; white-space: nowrap !important; padding: 0 2px; }"
---
stateDiagram-v2
[*] --> QUEUED : [webserver / daemon]<br/>①② run を作ってキューに積む
QUEUED --> STARTING : [daemon] ③ ECS RunTask で<br/>run worker を起動
STARTING --> STARTED : [run worker] 実行を開始
STARTING --> FAILURE : [daemon] 3 分(デフォルト値)以内に<br/>起動しない
STARTED --> SUCCESS : [run worker] 全 step が成功
STARTED --> FAILURE : [run worker] step が失敗
STARTED --> FAILURE : [daemon] run worker が<br/>落ちたのを検知
STARTED --> CANCELING : [daemon] 30 分を超えた
CANCELING --> FAILURE : [daemon] ECS タスクを<br/>停止して失敗に
SUCCESS --> [*]
FAILURE --> [*]
note right of STARTED
取り込みの一時的なエラーは
step 単位で再試行
(30 秒後・最大 3 回)。
run は STARTED のまま
end note
note right of FAILURE
どの線で FAILURE になっても
run_failure_sensor が
Slack に通知
end note
注意点 1: メモリ不足で run worker が落ちたとき
メモリ不足で OOM キラーに落とされた run worker は、ECS から見ると「ECS タスクの異常終了」、Dagster から見ると「run worker が落ちた」という 2 つの出来事になります。そのため、2 本の経路の両方で通知されます。
1 本目の経路: run worker が落ちたことを daemon が検知して(run monitoring 機能)、run を FAILURE にします。上の状態図の [daemon] run worker が落ちたのを検知、の線です
2 本目の経路: ECS タスクの異常終了として、EventBridge が拾ってくれます
Slack には 2 本の経路から 2 回通知されてしまいますが、OOM は見逃したくない失敗なので許容しています。
メモリ不足の場合は、run worker は自分で FAILURE を書き込めないので、run monitoring 機能を有効にしていないと、run は STARTED のまま残ってしまいます。
注意点 2: run がハングしたとき
前職では、run がハングして実行が終わらず、しかもそれに気づけないことがありました。
ハングした run はエラーを出さないので、何もしなければ失敗の通知も来ません。
そこで daemon に、30 分を超えた run を ECS タスクごと止めて FAILURE にさせています(状態図の STARTED → CANCELING → FAILURE)。FAILURE になるので、1 本目の経路で失敗通知が届きます。
30 分は私たちの非機能要件から決めた値で、dagster.yaml の run_monitoring.max_runtime_seconds で設定しています。
30 分を超えることが分かっているジョブは、dagster/max_runtime タグで上限を個別に変えられます。
注意点 3: daemon をデプロイしたとき
daemon や code-location が落ちると 1 本目の経路が止まるので、2 本目の EventBridge と CloudWatch アラームで気づけるようにしています。アラームは「1 分ごとに確認して、3 回続けて 0 台なら通知する」設定にしています。
ところが、daemon をデプロイするたびに、このアラームから通知が来ていました。そのたびに、daemon が落ちた要因を探すことになってしまいます。
Dagster は daemon を複数台で動かすことをサポートしていません(公式ドキュメント)。このため、daemon のデプロイでは古い ECS タスクを停止してから新しい ECS タスクを起動する設定にしています。code-location と webserver は、新しい ECS タスクを起動してから古い ECS タスクを停止するので、0 台になる時間がありません。
この入れ替えの間、daemon の ECS タスクは 0 台になります。計測すると、0 台の時間は約 3 分あることがわかりました。アラームの猶予もちょうど 3 分だったので、これが誤通知の原因でした。
調査の結果、daemon が短時間止まっている間に取りこぼしがあっても、問題はないことがわかりました。
スケジュール: daemon が止まっている間に実行時刻が来ても、daemon が戻ったときに実行してくれます(メンテナーの回答)
失敗通知のセンサー: 前回見たところから続きを見るので、止まっている間の失敗も通知されます
そこで、daemon だけアラームの猶予を 10 分に延ばすことで対処しました。
おわりに
Dagster on ECS で何が動いていて、失敗したらどう気づくのかを図で説明しました。
Dagster の仕組みや設定を把握できたら納得感を持ちつつ開発できると思いますので、その一助になれば幸いです。
次の記事を書く時は、取り込みのアセットを定義している Dagster の Component について書いていければ良いなと思っています。
付録: デフォルトから変えている Dagster の設定
この記事の図や本文に出てくる Dagster の設定のうち、デフォルトから変えているものです。再試行だけは取り込みのアセットのコードで、ほかは dagster.yaml で設定しています。
設定
デフォルト
この構成
デフォルトのままだと
storage
ローカルの SQLite
Aurora PostgreSQL
ECS サービス同士で run の状態を共有できない
compute_logs
ローカルのファイル
S3
run worker の ECS タスクが終わるとログが消える
run_launcher
DefaultRunLauncher
EcsRunLauncher
run が code-location のコンテナの中で動く
run_monitoring.enabled
無効
有効
異常終了した run が STARTED のまま残り、Dagster からの失敗通知も届かない
run_monitoring.max_runtime_seconds
0(上限なし)
1800(30 分)
ハングした run の ECS タスクが動き続ける
step の再試行(取り込みのみ)
なし
一時的なエラーなら 30 秒後・最大 3 回
接続エラー 1 回で run が FAILURE になる
Mentions
Is Dagster free? Self-hosting Dagster vs Dagster+ Yes, Dagster is free. The open-source core is Apache 2.0, and you can run all of it on your own servers without paying Dagster Labs anything. That answers the licence question. It doesn't answer the budget question. The budget question is whether you'd rather pay Dagster Labs or pay with your own hours. I self-host a production Dagster instance on a homelab Docker Swarm. It runs 112 assets, 57 schedules and 3 sensors for one person, me. Here is what the paid product costs today, what self-hosting has cost me in time, and a minimal setup you can start from. Is Dagster free? What open source gives you The open-source packages give you the whole orchestrator: assets, partitions, schedules, sensors, asset checks, the run queue and the web UI. Nothing in my pipeline needed a paid feature. The Dagster repository is licensed Apache-2.0. What you take on is running it. A self-hosted Dagster has four parts: 1. Postgres for run history, event logs and schedule state. SQLite works on a laptop; I use it for local development only. 2. Code locations: containers that load your definitions and serve them over gRPC. 3. dagster-webserver: the UI and the GraphQL API. 4. dagster-daemon: schedules, sensors, the run queue, retries and run monitoring. None of these is hard to start. Keeping them healthy for a year is the real work. What Dagster+ costs I checked dagster.io on 1 October 2026: | Plan | Price | Includes | Per credit | |---|---|---|---| | Solo | $10 / month | 1 user, 1 code location, 1 deployment | $0.040 | | Starter | $100 / month | up to 3 users, 5 code locations, 1 deployment | $0.035 | | Pro | contact sales | unlimited seats, code locations and deployments | custom | Serverless compute was $0.010 a minute on Solo and Starter. Solo and Starter had a 30-day free trial. The page defines a credit as one asset materialization or one op execution, and says a Hybrid deployment, where your own infrastructure runs the code, has no compute charge. There is no free tier on the page, only the trial. To see what credits mean for you, count materializations. My pipeline has 112 assets, 36 of them partitioned (34 daily). Every daily partition of every daily asset is a credit, every day, plus every retry. Do that sum for your own graph before you assume a plan fits. I haven't priced my pipeline on Dagster+, so I won't guess my number. --- This is the first part. The full post — including the rest of the working details — is on my site: Is Dagster free? Self-hosting Dagster vs Dagster+
図で理解する Dagster on ECS の構成と、失敗に気づく仕組み セーフィー株式会社のデータ分析基盤グループでエンジニアをしている渡部です。 セーフィーのデータ分析基盤では、データソースごとにばらばらだったデータ取り込み処理のやり方を揃えるため、Python ライブラリの dlt(data load tool)を採用することとなりました。 dlt は取り込みを実行するだけなので「いつ実行するか？」「失敗にどう気づくか？」の仕組みは別で用意する必要があります。 そこで、オーケストレーションツールとして、Dagster も合わせて導入することとなりました。 現在は、プロダクトの DB から Snowflake へデータを取り込む処理の一部を ECS 上の Dagster で動作させています。 この記事では、そんな Dagster を理解するために、ECS の構成と失敗に気づく仕組みを図を使って説明することを目指します。 ! 次のことはこの記事では扱いません。 アセット・スケジュール・センサーといった最も基本的な Dagster のコンセプト アセットを定義する Python コードの書き方(dagster.Component など) dlt の使い方 ! 内容は dagster 1.13.13、dagster-aws / dagster-dlt 0.29.13、dlt 1.29.0 の時点のものです。 ECS の構成 まずはインフラ全体の構成です。 図の Mermaid コード(再生成用)--- config: flowchart: wrappingWidth: 400 --- flowchart LR user@{ icon: "mdi:web", form: "square", label: "利用者のブラウザ", pos: "b", h: 48 } subgraph aws["AWS アカウント(データ分析基盤)"] waf@{ icon: "aws:waf", form: "square", label: "WAF", pos: "b", h: 48 } alb@{ icon: "aws:elb", form: "square", label: "ALB<br/>(Cognito で認証)", pos: "b", h: 48 } subgraph vpc["VPC"] subgraph ecs["ECS on Fargate"] web@{ icon: "aws:fargate", form: "square", label: "webserver<br/>(常駐)", pos: "b", h: 48 } daemon@{ icon: "aws:fargate", form: "square", label: "daemon<br/>(常駐)", pos: "b", h: 48 } cl@{ icon: "aws:fargate", form: "square", label: "code-location<br/>(常駐)", pos: "b", h: 48 } rw@{ icon: "aws:fargate", form: "square", label: "run worker<br/>(run ごとに起動)", pos: "b", h: 48 } end db@{ icon: "aws:aurora", form: "square", label: "Aurora PostgreSQL<br/>(Dagster DB)", pos: "b", h: 48 } re@{ icon: "aws:lattice", form: "square", label: "Resource Endpoint<br/>(ソース DB ごと)", pos: "b", h: 48 } nat@{ icon: "aws:nat", form: "square", label: "NAT Gateway", pos: "b", h: 48 } end s3@{ icon: "aws:s3", form: "square", label: "S3<br/>(compute logs)", pos: "b", h: 48 } end subgraph src["ソース側の AWS アカウント"] srcdb@{ icon: "aws:aurora", form: "square", label: "Aurora<br/>(ソース DB)", pos: "b", h: 48 } end subgraph saas["SaaS"] sf@{ icon: "logos:snowflake-icon", form: "square", label: "Snowflake", pos: "b", h: 48 } dbt@{ icon: "logos:dbt-icon", form: "square", label: "dbt Cloud", pos: "b", h: 48 } slack@{ icon: "logos:slack-icon", form: "square", label: "Slack", pos: "b", h: 48 } end user --> waf --> alb --> web web -- gRPC --> cl daemon -- gRPC --> cl daemon -- RunTask --> rw ecs -- 5432 --> db ecs --> s3 rw --> re -- VPC Lattice --> srcdb ecs --> nat nat --> sf nat --> dbt nat --> slack style aws fill:#ffffff,stroke:#E7157B,stroke-width:2px style src fill:#ffffff,stroke:#E7157B,stroke-width:2px style vpc fill:#ffffff,stroke:#8C4FFF,stroke-width:2px style ecs fill:#FFF8F0,stroke:#ED7100,stroke-width:1px,stroke-dasharray:5 5 style saas fill:#F7F7F7,stroke:#7D8998,stroke-width:1px ECS の構成は、公式ドキュメントにある構成例(Deploying Dagster to Amazon Web Services)に沿っていて、とってもシンプルです。 ECS の上で動いているものは以下の 4 種類となっています。 ECS サービス / ECS タスク 役割 起動コマンド code-location アセットやスケジュールを定義したコードを読み込み、webserver と daemon の問い合わせに答える dagster code-server start -h 0.0.0.0 -p 4000 -m data_integration.definitions daemon スケジュールやセンサーを評価して、run を起動する dagster-daemon run -w workspace.yaml webserver UI を提供する dagster-webserver -h 0.0.0.0 -p 3000 -w workspace.yaml run worker データ取り込みの処理 = run を 1 つ実行する dagster api execute_run <JSON>(EcsRunLauncher という Python クラスが面倒を見てくれる) このうち code-location・daemon・webserver は常に起動している(常駐している)ECS タスクです。 一方で run(ジョブの 1 回の実行)を受け持つ run worker はアドホックに立ち上がる ECS タスクです。 現在の dagster.yaml の設定では、Dagster DB に実行履歴などが保存されており、標準出力・標準エラー出力は S3 にファイルとしてアップロードされるようにしています。 利用者はロードバランサー経由で webserver にアクセスすれば、ブラウザでこれらを確認したり、アセットを手動で実行することができます。 取り込みを実行する run worker は、dlt を使って、以下のような流れでデータソースである Aurora PostgreSQL からデータを Snowflake に連携しています。 VPC Lattice の Resource Endpoint 経由で別 AWS アカウントにある Aurora PostgreSQL に接続してデータを取得 NAT Gateway から Snowflake にデータを書き込み 取り込みが終わったあとの変換は dbt Cloud のジョブに任せていて、その起動も Dagster から行っています。 Dagster で処理が実行されるまでに起きること Dagster で処理が実行されるまでは、次の図の① → ③の流れになっています。 図の Mermaid コード(再生成用)--- config: flowchart: wrappingWidth: 400 --- flowchart LR subgraph web["webserver(常駐)"] ui["UI / GraphQL"] end subgraph daemon["daemon(常駐)"] sch["スケジュールとセンサーを評価"] launcher["EcsRunLauncher"] end subgraph cl["code-location(常駐)"] defs["Definitions<br/>(アセット・ジョブ・スケジュール・センサー)"] end subgraph rw["run worker(run ごとに起動)"] step["アセットの実行"] end pg[("Dagster DB")] ui -- "① gRPC で定義を取得" --> defs sch -- "① gRPC で定義を取得" --> defs ui -- "② run をキューに積む" --> pg sch -- "② run をキューに積む" --> pg pg -- "キューから取り出す" --> launcher launcher -- "③ ECS RunTask で起動" --> step step -- "イベントを書き込み" --> pg Dagster では、処理の定義(アセットやジョブ = Python のコード)があるのに対して、ジョブの 1 回の実行を表すのが run です。 ここら辺の用語を正しく使わないと理解が難しくなるため、少し整理しておきます。 用語 定義か実行か 何か アセット 定義 作りたいデータ(テーブルなど)と、その作り方 ジョブ 定義 一緒に実行するアセットの組み合わせ run 実行 ジョブの 1 回の実行。状態とログを持つ step 実行 run の中の、アセット 1 つ分の実行 run が作られてから実行されるまでは、次の 3 段階(作る・積む・起動する)があることがわかります。 ① run を作る webserver か daemon が code-location に問い合わせて、run を作ります。 このとき、実行計画(どの step をどの順で実行するか)と Docker イメージが run に記録されます ② run をキューに積む 作った run を、Dagster DB に QUEUED の状態で書き込みます。 ここまでは UI から実行したときは webserver、スケジュールやセンサーのときは daemon が担当します。 ③ run を起動する 起動自体は daemon が行います。 キューから run を取り出して、run に記録されたイメージで run worker の ECS タスクを作ります(EcsRunLauncher) Dagster の挙動を把握するコツ Dagster はワークフローエンジンということもあって、解説記事は多くありません。しかも、いろいろな環境にデプロイできて、処理も Python で書けることであればなんでもできてしまいます。インフラ構成・設定・実装までを考えると自由度が高くとても複雑に見えます。 全体を一度に見てしまうと複雑に見えてしまうので 1 つずつ分解して見ることが大切です。 一番メインとなるのは ECS サービス / ECS タスクです。ECS の構成のテーブルに記載した、起動コマンドに目を向けてみてください。 webserver と daemon は単に Dagster で用意されているコマンドを実行しているだけであり、設定ファイルである dagster.yaml と workspace.yaml だけを考えれば良いです。 一方で自分たちの Python コードが関連してくるのは code-location と run worker だけなのです。 エラー・失敗に気づく仕組み ここまでで実行されるまでの流れを見てきました。では、失敗した時に気づけるようにするためにはどうすれば良いでしょうか？ 現在、成功時・失敗時ともに社内の Slack チャンネルに通知を行なっていますが、ここからはどのように失敗に気づけるようにしたかにフォーカスしていきたいと思います。 さまざまなデータソースと連携できる Dagster では多種多様な失敗が起き得ます。 私が実際に遭遇したもので思い返せるものでも、以下のようなものが挙げられます。 Python が例外を出してエラーになるもの メモリ不足で OOM キラーに叩き落とされるもの エラーにならずにハングしてずっと処理が続くもの daemon や code-location など、Dagster そのものが落ちるもの Python の例外ならば、Dagster 内部からの監視・通知でカバーできますが、Dagster それ自体が落ちてしまうと失敗に気づくことすらできません。 そこで、次の図のように監視・通知の経路は 2 本作成してあります。 図の Mermaid コード(再生成用)--- config: flowchart: wrappingWidth: 400 --- flowchart LR subgraph dagster["1 本目の経路: Dagster が通知する(daemon と code-location が動いている間だけ)"] daemon@{ icon: "aws:fargate", form: "square", label: "daemon", pos: "b", h: 48 } cl@{ icon: "aws:fargate", form: "square", label: "code-location<br/>(run_failure_sensor)", pos: "b", h: 48 } daemon -- 定期的に呼び出し --> cl end subgraph aws["2 本目の経路: AWS が通知する(Dagster が落ちていても通知される)"] ecsfail@{ icon: "aws:ecs", form: "square", label: "ECS タスクの<br/>異常終了・起動の失敗", pos: "b", h: 48 } eb@{ icon: "aws:eventbridge", form: "square", label: "EventBridge", pos: "b", h: 48 } zero@{ icon: "aws:ecs", form: "square", label: "常駐の ECS タスクが<br/>0 台のまま", pos: "b", h: 48 } cw@{ icon: "aws:cloudwatch", form: "square", label: "CloudWatch アラーム", pos: "b", h: 48 } sns@{ icon: "aws:sns", form: "square", label: "SNS", pos: "b", h: 48 } q@{ icon: "aws:q", form: "square", label: "Amazon Q Developer", pos: "b", h: 48 } ecsfail --> eb --> sns zero --> cw --> sns sns --> q end slack@{ icon: "logos:slack-icon", form: "square", label: "Slack", pos: "b", h: 48 } cl -- Incoming Webhook --> slack q --> slack style dagster fill:#ffffff,stroke:#7D8998,stroke-width:1px style aws fill:#ffffff,stroke:#E7157B,stroke-width:2px 1 本目の経路は、Dagster 内部からのものです。run_failure_sensor センサーを使用しています。 センサーの定期実行を支えているのは daemon、センサーの本体の判定ロジックの Python が動くのは code-location です。 内部からの監視・通知なので、daemon か code-location が落ちていると、失敗の通知が届きません。 2 本目の経路では、Dagster の外からの通知を用意しました。こちらは AWS の機能を使っています。 EventBridge: ECS タスクの異常終了と起動の失敗を検知する CloudWatch アラーム: 常駐サービスの ECS タスクが 0 台になったことを検知する どちらの通知も、SNS から Amazon Q Developer(旧 AWS Chatbot)を経由して Slack に通知されます。 以上、どの失敗をどちらの経路で拾うかをまとめると、次のようになります。 失敗 気づく経路 Python の例外 1 本目 メモリ不足(OOM) 2 本目(EventBridge)と 1 本目の両方 ハング 1 本目(30 分で FAILURE にする) Dagster そのものが落ちる 2 本目(EventBridge と CloudWatch アラーム) 運用して分かった注意点 ここまでが失敗に気づく仕組みです。ここからは、運用してみて分かった注意点を 3 つほど書かせてもらいます。 どの注意点も run の状態の変わり方が関わるので、先に図にしておきます。 図の Mermaid コード(再生成用)--- config: themeCSS: "foreignObject { overflow: visible; } foreignObject div { max-width: none !important; white-space: nowrap !important; padding: 0 2px; }" --- stateDiagram-v2 [*] --> QUEUED : [webserver / daemon]<br/>①② run を作ってキューに積む QUEUED --> STARTING : [daemon] ③ ECS RunTask で<br/>run worker を起動 STARTING --> STARTED : [run worker] 実行を開始 STARTING --> FAILURE : [daemon] 3 分(デフォルト値)以内に<br/>起動しない STARTED --> SUCCESS : [run worker] 全 step が成功 STARTED --> FAILURE : [run worker] step が失敗 STARTED --> FAILURE : [daemon] run worker が<br/>落ちたのを検知 STARTED --> CANCELING : [daemon] 30 分を超えた CANCELING --> FAILURE : [daemon] ECS タスクを<br/>停止して失敗に SUCCESS --> [*] FAILURE --> [*] note right of STARTED 取り込みの一時的なエラーは step 単位で再試行 (30 秒後・最大 3 回)。 run は STARTED のまま end note note right of FAILURE どの線で FAILURE になっても run_failure_sensor が Slack に通知 end note 注意点 1: メモリ不足で run worker が落ちたとき メモリ不足で OOM キラーに落とされた run worker は、ECS から見ると「ECS タスクの異常終了」、Dagster から見ると「run worker が落ちた」という 2 つの出来事になります。そのため、2 本の経路の両方で通知されます。 1 本目の経路: run worker が落ちたことを daemon が検知して(run monitoring 機能)、run を FAILURE にします。上の状態図の [daemon] run worker が落ちたのを検知、の線です 2 本目の経路: ECS タスクの異常終了として、EventBridge が拾ってくれます Slack には 2 本の経路から 2 回通知されてしまいますが、OOM は見逃したくない失敗なので許容しています。 メモリ不足の場合は、run worker は自分で FAILURE を書き込めないので、run monitoring 機能を有効にしていないと、run は STARTED のまま残ってしまいます。 注意点 2: run がハングしたとき 前職では、run がハングして実行が終わらず、しかもそれに気づけないことがありました。 ハングした run はエラーを出さないので、何もしなければ失敗の通知も来ません。 そこで daemon に、30 分を超えた run を ECS タスクごと止めて FAILURE にさせています(状態図の STARTED → CANCELING → FAILURE)。FAILURE になるので、1 本目の経路で失敗通知が届きます。 30 分は私たちの非機能要件から決めた値で、dagster.yaml の run_monitoring.max_runtime_seconds で設定しています。 30 分を超えることが分かっているジョブは、dagster/max_runtime タグで上限を個別に変えられます。 注意点 3: daemon をデプロイしたとき daemon や code-location が落ちると 1 本目の経路が止まるので、2 本目の EventBridge と CloudWatch アラームで気づけるようにしています。アラームは「1 分ごとに確認して、3 回続けて 0 台なら通知する」設定にしています。 ところが、daemon をデプロイするたびに、このアラームから通知が来ていました。そのたびに、daemon が落ちた要因を探すことになってしまいます。 Dagster は daemon を複数台で動かすことをサポートしていません(公式ドキュメント)。このため、daemon のデプロイでは古い ECS タスクを停止してから新しい ECS タスクを起動する設定にしています。code-location と webserver は、新しい ECS タスクを起動してから古い ECS タスクを停止するので、0 台になる時間がありません。 この入れ替えの間、daemon の ECS タスクは 0 台になります。計測すると、0 台の時間は約 3 分あることがわかりました。アラームの猶予もちょうど 3 分だったので、これが誤通知の原因でした。 調査の結果、daemon が短時間止まっている間に取りこぼしがあっても、問題はないことがわかりました。 スケジュール: daemon が止まっている間に実行時刻が来ても、daemon が戻ったときに実行してくれます(メンテナーの回答) 失敗通知のセンサー: 前回見たところから続きを見るので、止まっている間の失敗も通知されます そこで、daemon だけアラームの猶予を 10 分に延ばすことで対処しました。 おわりに Dagster on ECS で何が動いていて、失敗したらどう気づくのかを図で説明しました。 Dagster の仕組みや設定を把握できたら納得感を持ちつつ開発できると思いますので、その一助になれば幸いです。 次の記事を書く時は、取り込みのアセットを定義している Dagster の Component について書いていければ良いなと思っています。 付録: デフォルトから変えている Dagster の設定 この記事の図や本文に出てくる Dagster の設定のうち、デフォルトから変えているものです。再試行だけは取り込みのアセットのコードで、ほかは dagster.yaml で設定しています。 設定 デフォルト この構成 デフォルトのままだと storage ローカルの SQLite Aurora PostgreSQL ECS サービス同士で run の状態を共有できない compute_logs ローカルのファイル S3 run worker の ECS タスクが終わるとログが消える run_launcher DefaultRunLauncher EcsRunLauncher run が code-location のコンテナの中で動く run_monitoring.enabled 無効 有効 異常終了した run が STARTED のまま残り、Dagster からの失敗通知も届かない run_monitoring.max_runtime_seconds 0(上限なし) 1800(30 分) ハングした run の ECS タスクが動き続ける step の再試行(取り込みのみ) なし 一時的なエラーなら 30 秒後・最大 3 回 接続エラー 1 回で run が FAILURE になる