The Complete Production SOP: How to Deploy and Run Dagster Software-Defined Assets & Data Lineage in Autonomous AI Agents

---
name: "dagster-software-defined-assets"
description: "Declare data pipelines as Software-Defined Assets with integrated column-level lineage."
metadata:
  author: "Dagster Labs"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "create dagster asset"
  - "setup dagster partitions"
  - "configure dagster io manager"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# Dagster Software-Defined Assets & Data Lineage

Implements data platforms using Dagster: declaring assets with @asset, defining upstream dependencies, partition definitions (DailyPartitionsDefinition), I/O managers, and declarative freshness policies.

## When to Invoke
Invoke this skill whenever you need to:
- create dagster asset
- setup dagster partitions
- configure dagster io manager

## Execution SOP & Rules
1. Define assets using @asset(deps=[upstream_asset]) or accept upstream assets directly as function parameters.
2. Assign partitions_def = DailyPartitionsDefinition(start_date='2025-01-01') for date-partitioned tables.
3. Declare auto-materialize policies with AutoMaterializePolicy.eager() for real-time asset freshness.
4. Launch the Dagit/Dagster web UI to inspect the asset graph, execution history, and data lineage.

from dagster import Definitions, load_assets_from_modules, EnvVar
from dagster_duckdb import DuckDBIOManager
from . import assets

all_assets = load_assets_from_modules([assets])

defs = Definitions(
    assets=all_assets,
    resources={
        "io_manager": DuckDBIOManager(
            database=EnvVar("DUCKDB_DATABASE_PATH").get_value("warehouse.duckdb")
        )
    }
)

import pandas as pd
from dagster import asset, Output, MetadataValue, DailyPartitionsDefinition, AssetExecutionContext

DAILY_PARTITION = DailyPartitionsDefinition(start_date="2025-01-01")

@asset(
    partitions_def=DAILY_PARTITION,
    group_name="ingestion",
    compute_kind="python"
)
def raw_telemetry_events(context: AssetExecutionContext) -> Output[pd.DataFrame]:
    partition_date = context.partition_key
    context.log.info(f"Processing partition: {partition_date}")
    
    # Ingest partition data
    df = pd.DataFrame({
        "event_id": [f"{partition_date}_1", f"{partition_date}_2"],
        "user_id": [101, 102],
        "duration_ms": [342.5, 120.1]
    })
    
    return Output(
        value=df,
        metadata={
            "record_count": MetadataValue.int(len(df)),
            "partition_window": MetadataValue.text(partition_date),
            "columns": MetadataValue.json({"fields": list(df.columns)})
        }
    )

@asset(
    partitions_def=DAILY_PARTITION,
    group_name="analytics",
    compute_kind="python"
)
def daily_user_aggregates(raw_telemetry_events: pd.DataFrame) -> Output[pd.DataFrame]:
    aggregated = raw_telemetry_events.groupby("user_id").agg({
        "duration_ms": ["sum", "mean"]
    }).reset_index()
    aggregated.columns = ["user_id", "total_duration", "mean_duration"]
    
    return Output(
        value=aggregated,
        metadata={
            "user_count": MetadataValue.int(len(aggregated)),
            "schema": MetadataValue.md("| column | type |\n|---|---|\n| user_id | int |\n| total_duration | float |")
        }
    )

#!/usr/bin/env bash
set -euo pipefail

echo "Validating Dagster Definitions..."
dagster definitions validate -f analytics_pipeline/definitions.py

echo "Running Ruff lint checks..."
ruff check analytics_pipeline/

echo "Dagster Software-Defined Assets verified successfully."