Orchestrate Data Pipelines with Dagster

This skill provides expertise in Dagster, the modern data orchestration platform, with deep knowledge of its asset-oriented approach, Software-Defined Assets (SDAs), jobs, ops, resources, sensors, schedules, and the broader Dagster ecosystem for building production-ready data pipelines. Core Dagster principles emphasize asset-oriented development - thinking in terms of data assets rather than just tasks, modeling data dependencies explicitly through asset lineage, using SDAs as the primary abstraction, leveraging asset materialization for incremental computation, and designing assets to be idempotent and testable - alongside declarative pipeline definition, which specifies what should exist rather than just how to compute it, using type annotations and metadata for self-documenting pipelines, and separating business logic from orchestration concerns.

Software-Defined Assets are demonstrated with a basic asset chain (an @asset-decorated extraction function, a dependent cleaning asset declared via deps, and a summary asset consuming a named input with attached metadata like owner and SLA minutes) and multi-assets for related outputs - a single function producing bronze/silver/gold layered outputs (raw ingestion, cleaned/validated data, and business-ready aggregations) via @multi_asset with named AssetOut outputs. Resources and configuration cover ConfigurableResource classes for database connections (SQLAlchemy engine with configurable connection string and pool size) and S3 access (boto3 client with credentials), injected directly into assets as typed parameters.

Jobs, schedules, and sensors cover defining a job from a selection of assets with resource configuration, a cron-based ScheduleDefinition (e.g. running daily at 2 AM), and a file-based sensor that polls a directory for new files and yields a RunRequest with a unique run key and resource configuration when new data arrives. Partitioned assets cover time-based partitioning - a DailyPartitionsDefinition applied to an asset so each run processes a specific date partition, with the partition key driving a parameterized query for that day's data.

Use this skill when building or extending a Dagster data pipeline - defining Software-Defined Assets, wiring up resources, scheduling or sensor-triggering jobs, or partitioning assets by time. It is well suited to data engineering teams adopting Dagster's asset-centric orchestration model. It is not meant for non-Dagster orchestration (Airflow, Prefect), or for simple scripts with no meaningful asset lineage or scheduling need.

Input: the data sources, transformation logic, and scheduling/partitioning requirements for the pipeline.

Output: Dagster asset definitions, resource configurations, jobs with schedules or sensors, and partitioned assets. Example basic asset chain:

@asset
def raw_orders() -> pd.DataFrame:
    return pd.read_csv("orders.csv")

@asset(deps=[raw_orders])
def cleaned_orders(raw_orders: pd.DataFrame) -> pd.DataFrame:
    return raw_orders.dropna().reset_index(drop=True)

Builds on the Dagster framework, integrating with SQLAlchemy for database resources, boto3 for S3, and pandas for data transformation within assets.

Data engineers building or extending Dagster pipelines, and teams adopting asset-oriented orchestration who need proper resource management, scheduling, and time-based partitioning.