Dagster 자산 구축하기
사용 시점
- Dagster 소프트웨어 정의 자산 및 작업을 생성하거나 리팩토링할 때
- 테이블/파일/ML 모델을 계보(lineage)를 포함한 자산으로 모델링할 때
- 파티션, 백필, 자산 검사, 스케줄 또는 센서를 추가할 때
- Airflow에는 사용하지 마십시오 (Airflow 스킬을 사용하세요).
워크플로우
- [ ] 각 출력을 @asset으로 모델링하고, 함수 인자를 통해 의존성 선언
- [ ] 시간/카테고리별 데이터에 파티션 추가
- [ ] IO(읽기/쓰기)를 IO 매니저 또는 리소스로 이동
- [ ] 데이터 품질을 위한 자산 검사 추가
- [ ] 스케줄/센서를 통해 구체화(materialize)
- 자산 중심으로 생각하고, 태스크 중심이 아닙니다. 자산은 영구적인 객체(테이블, 파일, 모델)입니다. 함수 파라미터로 업스트림 자산을 참조하여 의존성을 선언하면 Dagster가 자동으로 계보 그래프를 구축합니다.
- 파티션은 자연스럽게 분할되는(일별, 지역별) 자산에 적용하여 한 번에 하나의 조각만 구체화하거나 백필할 수 있게 합니다.
- 리소스 및 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 매니저를 사용하세요.
- 비멱등성 파티셔닝 자산 — 파티션을 구체화할 때는 추가(append)가 아니라 덮어쓰기(overwrite)해야 합니다 (
writing-idempotent-transformations참조). - 자산 검사 건너뛰기 — 검사가 없으면 잘못된 데이터가 조용히 구체화됩니다. Dagster는 UI에서 검사 실패를 표시하고 다운스트림을 차단할 수 있습니다.
- 의존 자산 간 파티션 정의 불일치 —
partitions_def를 일관되게 유지하여 매핑이 올바르게 작동하도록 하세요.