Dagsterアセットの構築 は何をしますか？
使用するタイミング
- Dagsterのソフトウェア定義アセットやジョブを作成・リファクタリングするとき
- テーブル/ファイル/MLモデルをリネージ付きのアセットとしてモデル化するとき
- パーティション、バックフィル、アセットチェック、スケジュール、センサーを追加するとき
- Airflowには使用しないでください（Airflow用のスキルを使用してください）。
ワークフロー
- [ ] 各出力を @asset としてモデル化し、関数引数で deps を宣言する
- [ ] 時間/カテゴリでスライスされたデータにパーティションを追加する
- [ ] IO（読み取り/書き込み）をIOマネージャーまたはリソースに移動する
- [ ] データ品質のためのアセットチェックを追加する
- [ ] スケジュール/センサーでマテリアライズする
- タスクではなくアセットで考える。 アセットとは永続的なオブジェクト（テーブル、ファイル、モデル）です。上流のアセットを関数パラメーターとして参照することで依存関係を宣言します。Dagsterが自動的にリネージグラフを構築します。
- パーティション は、自然にスライスされる（日次、地域など）アセットに設定し、一度に1スライスずつマテリアライズ/バックフィルできるようにします。
- リソースとIOマネージャー は接続と読み書きロジックを保持し、アセット本体を変換処理に集中させ、テスト可能にします。
- アセットチェック はアセットにデータ品質のアサーションを付加します。
パターン
依存関係を持つパーティション分割されたアセット:
import dagster as dg
daily = dg.DailyPartitionsDefinition(start_date="2026-01-01")
@dg.asset(partitions_def=daily)
def raw_orders(context: dg.AssetExecutionContext) -> None:
day = context.partition_key
write_parquet(f"raw/orders/{day}.parquet", fetch_orders(day))
@dg.asset(partitions_def=daily)
def orders_clean(context, raw_orders) -> None: # raw_orders に依存
day = context.partition_key
transform_and_load(day)
@dg.asset_check(asset=orders_clean)
def no_null_ids(context) -> dg.AssetCheckResult:
n = count_null_order_ids(context.partition_key)
return dg.AssetCheckResult(passed=n == 0, metadata={"null_ids": n})
リソース / IOマネージャー — アセット内部で構築する代わりにウェアハウスクライアントを注入:
@dg.asset
def orders_summary(context, warehouse: WarehouseResource, orders_clean):
warehouse.execute("insert into summary select ...")
スケジュール でパーティション分割されたジョブを設定し、各実行で最新のパーティションをマテリアライズします。センサー を使用すると、上流のファイルやアセットが出現したときにマテリアライズできます。
よくある落とし穴
- タスク思考 — すべてに生の
@op/ジョブを使うと、アセットが無料で提供するリネージ、可観測性、パーティション対応バックフィルが失われます。 - アセット内部でのIO — 接続をハードコードするとアセットがテスト不可能になります。リソース/IOマネージャーを使用してください。
- 非冪等なパーティション分割アセット — パーティションのマテリアライズはそのパーティションを上書きする必要があり、追記してはいけません（
writing-idempotent-transformationsを参照）。 - アセットチェックのスキップ — チェックがないと、不良データがサイレントにマテリアライズされます。DagsterはUIでチェックの失敗を表示し、下流をブロックできます。
- 依存アセット間のパーティション定義の不一致 —
partitions_defを一貫させ、マッピングが正しく解決されるようにしてください。