Dagster 실무 스터디 가이드

1. Dagster란?

Dagster는 데이터 애플리케이션을 구축, 실행, 모니터링하기 위한 클라우드 네이티브 데이터 오케스트레이션 플랫폼입니다. 기존의 DAG 스케줄러를 넘어 Asset(자산)을 중심으로 한 개발 경험과 데이터 품질 관리를 제공합니다.

2. 왜 써야 하는가?

Airflow, Prefect, Luigi 등과 비교한 Dagster의 차별점:

3. 핵심 개념

주요 컴포넌트

• Asset: 데이터 자산(테이블, 파일, 모델)을 표현하는 객체. 데이터 계보의 중심.
• Op: 실제 작업을 수행하는 함수 단위. 순수 함수 권장.
• Job: Ops/Assets를 실행 가능한 단위로 묶은 것.
• Resource: 데이터베이스 연결, API 클라이언트 등 외부 의존성.
• I/O Manager: Asset의 로드/저장을 관리하는 저장소 추상화.
• Sensor: 파일 생성, API 응답 등 외부 이벤트를 감지하여 Job 실행.

4. 퀵스타트

4.1 설치

# pip를 이용한 설치 (최신 버전)
pip install dagster dagster-webserver

# 또는 poetry 사용
poetry add dagster dagster-webserver

# 추가 패키지 (데이터 처리용)
pip install pandas dagster-pandas

4.2 첫 번째 Asset 정의

# my_first_dagster.py
from dagster import asset, Definitions
import pandas as pd

@asset
def raw_data() -> pd.DataFrame:
    """원본 데이터를 생성하는 Asset"""
    return pd.DataFrame({"id": [1, 2, 3], "value": [10, 20, 30]})

@asset
def processed_data(raw_data: pd.DataFrame) -> pd.DataFrame:
    """raw_data를 가공하는 Asset"""
    df = raw_data.copy()
    df["value_doubled"] = df["value"] * 2
    return df

@asset
def summary_report(processed_data: pd.DataFrame) -> str:
    """최종 리포트 생성 Asset"""
    total = processed_data["value_doubled"].sum()
    return f"Total doubled value: {total}"

defs = Definitions(
    assets=[raw_data, processed_data, summary_report]
)

4.3 실행 및 UI 확인

# Dagster UI 서버 실행
dagster dev -f my_first_dagster.py

# 브라우저에서 http://localhost:3000 접속
# Asset 그래프 확인 및 수동 실행 가능

5. 핵심 설정 레퍼런스

설정 파일 예시 (dagster.yaml)

# dagster.yaml
storage:
  postgres:
    postgres_db:
      username: { env: DAGSTER_PG_USERNAME }
      password: { env: DAGSTER_PG_PASSWORD }
      hostname: { env: DAGSTER_PG_HOST }
      db_name: { env: DAGSTER_PG_DB }
      port: 5432

run_storage:
  module: dagster_postgres.run_storage
  class: PostgresRunStorage
  config:
    postgres_url: { env: DAGSTER_PG_URL }

schedule_storage:
  module: dagster_postgres.schedule_storage
  class: PostgresScheduleStorage
  config:
    postgres_url: { env: DAGSTER_PG_URL }

scheduler:
  module: dagster.core.scheduler
  class: DagsterDaemonScheduler

run_launcher:
  module: dagster.core.launcher
  class: DefaultRunLauncher

6. 실무 팁 & 트러블슈팅

from dagster import resource

@resource
def database_connection():
    return create_engine("postgresql://...")

디버깅 명령어

# 정의 로드 확인
dagster job list -f my_pipeline.py

# 특정 Job의 실행 계획 확인
dagster job preview -j my_job -f my_pipeline.py

# UI 없이 실행 (CLI)
dagster job execute -j my_job -f my_pipeline.py

7. 참고 자료

• 📘 공식 문서 (한글 부분 번역 포함) - 개념, 튜토리얼, API 레퍼런스