Dagster: 에셋 기반 데이터 파이프라인 오케스트레이터 — 2026 프로덕션 구축 가이드

from dagster import asset, Definitions
import pandas as pd

@asset(key="raw_customers")
def raw_customers():
    """Load raw customer data from upstream CSV."""
    df = pd.read_csv("s3://data-lake/raw/customers.csv")
    return df

@asset(key="cleaned_customers")
def cleaned_customers(raw_customers):
    """Clean and deduplicate customer records."""
    df = raw_customers.drop_duplicates(subset="email")
    df["email"] = df["email"].str.lower().str.strip()
    return df

@asset(key="customer_metrics")
def customer_metrics(cleaned_customers):
    """Aggregate customer metrics for reporting."""
    return cleaned_customers.groupby("country").agg(
        total_customers=("customer_id", "count"),
        avg_lifetime_value=("ltv", "mean")
    ).reset_index()

# Define the repository of assets
defs = Definitions(assets=[raw_customers, cleaned_customers, customer_metrics])

from dagster import AssetSelection, define_asset_job, ScheduleDefinition

# Run daily at 6 AM UTC
daily_job = define_asset_job(
    name="daily_customer_pipeline",
    selection=AssetSelection.all()
)

daily_schedule = ScheduleDefinition(
    job=daily_job,
    cron_schedule="0 6 * * *",  # 6 AM UTC daily
    default_status=DefaultScheduleStatus.RUNNING
)

from dagster import AutoMaterializePolicy

@asset(
    auto_materialize_policy=AutoMaterializePolicy.eager()
)
def customer_metrics(cleaned_customers):
    """Automatically rebuilds whenever upstream data changes."""
    return cleaned_customers.groupby("country").agg(...)

from dagster import asset_check, AssetCheckResult

@asset_check(asset=raw_customers)
def no_empty_customers(raw_customers):
    """Validate that customer table is not empty."""
    row_count = len(raw_customers)
    return AssetCheckResult(
        passed=row_count > 0,
        metadata={"row_count": row_count}
    )

@asset_check(asset=cleaned_customers)
def unique_emails(cleaned_customers):
    """Validate email uniqueness after deduplication."""
    duplicate_count = cleaned_customers["email"].duplicated().sum()
    return AssetCheckResult(
        passed=duplicate_count == 0,
        metadata={"duplicate_emails": duplicate_count}
    )

# Create a virtual environment
python -m venv .venv
source .venv/bin/activate

# Install Dagster and the webserver
pip install dagster dagster-webserver dagster-graphql

# Verify installation
dagster --version
# dagster, version 1.13.2

# Install the dg CLI tool
pip install dagster-dg

# Scaffold a new project
dg scaffold project my_data_platform --python-version 3.11
cd my_data_platform

# The scaffold creates:
# my_data_platform/
# ├── my_data_platform/
# │   ├── __init__.py
# │   ├── definitions.py
# │   └── assets.py
# ├── pyproject.toml
# └── setup.py

# my_data_platform/assets.py
from dagster import asset, Definitions
import pandas as pd

@asset
def hello_world():
    """First asset: creates a sample dataset."""
    return pd.DataFrame({
        "name": ["Alice", "Bob", "Charlie"],
        "score": [85, 92, 78]
    })

defs = Definitions(assets=[hello_world])

# From the project root
dagster dev -h 0.0.0.0 -p 3000

# docker-compose.yml
version: "3.8"
services:
  dagster-postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_USER: dagster
      POSTGRES_PASSWORD: dagster
      POSTGRES_DB: dagster
    volumes:
      - postgres_data:/var/lib/postgresql/data

  dagster-daemon:
    build: .
    command: dagster-daemon run
    environment:
      DAGSTER_POSTGRES_USER: dagster
      DAGSTER_POSTGRES_PASSWORD: dagster
      DAGSTER_POSTGRES_DB: dagster
      DAGSTER_POSTGRES_HOST: dagster-postgres
    depends_on:
      - dagster-postgres

  dagster-webserver:
    build: .
    command: dagster-webserver -h 0.0.0.0 -p 3000
    ports:
      - "3000:3000"
    environment:
      DAGSTER_POSTGRES_USER: dagster
      DAGSTER_POSTGRES_PASSWORD: dagster
      DAGSTER_POSTGRES_DB: dagster
      DAGSTER_POSTGRES_HOST: dagster-postgres
    depends_on:
      - dagster-postgres

volumes:
  postgres_data:

docker-compose up --build -d

# Integrate dbt models as Dagster assets
from dagster_dbt import DbtProject, dbt_assets
from dagster import AssetExecutionContext

dbt_project = DbtProject(
    project_dir="./dbt_project",
    profiles_dir="./dbt_project/profiles"
)

@dbt_assets(manifest=dbt_project.manifest_path)
def dbt_models(context: AssetExecutionContext, dbt: DbtCliResource):
    """Every dbt model becomes a Dagster asset automatically."""
    yield from dbt.cli(["build"], context=context).stream()

from dagster_snowflake import SnowflakeResource
from dagster import asset, Definitions

@asset
def snowflake_raw_orders(context, snowflake: SnowflakeResource):
    """Query raw orders from Snowflake."""
    with snowflake.get_connection() as conn:
        return conn.execute("SELECT * FROM RAW.ORDERS").fetch_pandas_all()

defs = Definitions(
    assets=[snowflake_raw_orders],
    resources={
        "snowflake": SnowflakeResource(
            account="xyz123",
            user="ETL_USER",
            password={"env": "SNOWFLAKE_PASSWORD"},
            database="ANALYTICS",
            warehouse="ETL_WH"
        )
    }
)

from dagster_airbyte import AirbyteResource, sync_assets

airbyte = AirbyteResource(
    host="localhost",
    port="8000",
    username="airbyte",
    password={"env": "AIRBYTE_PASSWORD"}
)

# Generate assets from Airbyte connections
airbyte_assets = sync_assets(
    connection_id="123e4567-e89b-12d3-a456-426614174000",
    airbyte=airbyte
)

from dagster import DailyPartitionsDefinition, asset

daily_partition = DailyPartitionsDefinition(start_date="2024-01-01")

@asset(partitions_def=daily_partition)
def daily_sales(context):
    """Process one day of sales data per partition."""
    partition_date = context.partition_key
    query = f"SELECT * FROM sales WHERE date = '{partition_date}'"
    return run_query(query)

# Backfill 30 days with one command
dagster asset backfill -p daily_sales --from 2024-01-01 --to 2024-01-30

# resources.py — different configs for dev/staging/prod
from dagster_snowflake import SnowflakeResource

snowflake_dev = SnowflakeResource(
    account="xyz123",
    user="dev_user",
    password={"env": "SNOWFLAKE_DEV_PASSWORD"},
    database="dev_analytics",
    schema="public",
    warehouse="DEV_WH"
)

snowflake_prod = SnowflakeResource(
    account="xyz123",
    user="prod_etl",
    password={"env": "SNOWFLAKE_PROD_PASSWORD"},
    database="prod_analytics",
    schema="public",
    warehouse="PROD_WH_LARGE"
)

from dagster import sensor, RunRequest
from dagster_slack import make_slack_on_run_failure_sensor

# Alert on any run failure
slack_failure_sensor = make_slack_on_run_failure_sensor(
    channel="#data-alerts",
    slack_token={"env": "SLACK_BOT_TOKEN"},
    text_fn=lambda context: (
        f"Pipeline failed: {context.pipeline_run.job_name}\n"
        f"Run ID: {context.pipeline_run.run_id}\n"
        f"Error: {context.failure_event.message}"
    )
)

# Sensor that triggers when a new file arrives in S3
@sensor(job=daily_customer_pipeline)
def s3_file_sensor():
    new_files = check_s3_for_new_files("s3://data-lake/incoming/")
    for file in new_files:
        yield RunRequest(
            run_key=file.etag,
            run_config={"ops": {"raw_customers": {"config": {"s3_path": file.key}}}}
        )

# workspace.yaml
load_from:
  - python_module:
      module_name: analytics_team.definitions
      location_name: analytics
  - python_module:
      module_name: ml_team.definitions
      location_name: ml_platform
  - python_module:
      module_name: finance_team.definitions
      location_name: finance

# 1. Provision a Droplet with Docker pre-installed
# Use my referral link for $200 free credit:
# https://m.do.co/c/eca87ac14ee0

# 2. Clone your Dagster project
git clone https://github.com/your-org/dagster-platform.git
cd dagster-platform

# 3. Launch with Docker Compose
docker-compose -f docker-compose.prod.yml up -d

# 4. Verify health
curl http://your-droplet-ip:3000/health

# docker-compose.prod.yml
version: "3.8"
services:
  dagster-webserver:
    image: your-registry/dagster-platform:latest
    restart: always
    ports:
      - "3000:3000"
    environment:
      - DAGSTER_HOME=/opt/dagster/dagster_home
    volumes:
      - dagster_home:/opt/dagster/dagster_home

  dagster-daemon:
    image: your-registry/dagster-platform:latest
    restart: always
    command: dagster-daemon run
    volumes:
      - dagster_home:/opt/dagster/dagster_home

  postgres:
    image: postgres:15-alpine
    restart: always
    environment:
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  dagster_home:
  postgres_data:

from dagster import materialize

def test_customer_metrics():
    mock_customers = pd.DataFrame({
        "country": ["US", "US", "UK"],
        "customer_id": [1, 2, 3],
        "ltv": [100.0, 200.0, 150.0]
    })
    result = materialize(
        [customer_metrics],
        resources={"cleaned_customers": mock_customers}
    )
    assert result.success
    metrics = result.output_for_node("customer_metrics")
    assert len(metrics) == 2  # US and UK