建構 Dagster 資產
使用時機
- 建立或重構 Dagster 的軟體定義資產與任務。
- 將表格/檔案/ML 模型建模成具備血緣關係的資產。
- 新增分區、回填、資產檢查、排程或感測器。
- 不要用於 Airflow（請使用 Airflow 專用技能）。
工作流程
- [ ] 將每個輸出建模為 @asset；透過函數參數聲明依賴
- [ ] 為時間/類別切分的數據新增分區
- [ ] 將 IO（讀取/寫入）移至 IO Manager 或資源中
- [ ] 新增資產檢查以確保資料品質
- [ ] 使用排程/感測器來產生物質化
- 以資產為思考中心，而非任務。 資產是一個持久化的物件（資料表、檔案、模型）。透過將上游資產作為函數參數來聲明依賴關係——Dagster 會自動建立血緣圖。
- 為自然可切分的資產（按天、按區域）新增分區，這樣你就可以一次產生物質化或回填一個切片。
- 資源和 IO Manager 負責管理連線與讀寫邏輯，使資產主體專注於轉換邏輯，並讓資產更容易測試。
- 資產檢查 將資料品質斷言附加到資產上。
模式
帶有分區的資產及其依賴：
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 Manager — 注入倉庫客戶端，而非在資產內部建立：
@dg.asset
def orders_summary(context, warehouse: WarehouseResource, orders_clean):
warehouse.execute("insert into summary select ...")
排程一個分區任務，使得每次執行自動產生物質化最新分區；使用感測器在上游檔案/資產出現時觸發產生物質化。
常見陷阱
- 任務思考 — 到處使用原生的
@op/jobs 會失去資產提供的血緣、可觀測性以及分區感知的回填功能。 - 在資產函數體內進行 IO — 硬編碼連線使資產難以測試；應使用資源/IO Manager。
- 非冪等的分區資產 — 產生物質化一個分區時應覆蓋該分區，而不是追加（參見
writing-idempotent-transformations）。 - 跳過資產檢查 — 不進行檢查時，壞數據會靜默地產生物質化；Dagster 會在 UI 中顯示檢查失敗並可阻止下游。
- 依賴的資產之間分區定義不匹配 — 確保
partitions_def一致，以便對應關係能正確解析。