Orchestrate Data Pipelines with Dagster

Dagster Pipeline Expert Agent is a prompt for building data pipelines around software-defined assets. It emphasizes explicit dependencies and lineage, idempotent and testable computations, type annotations, metadata, and separation of business logic from orchestration. Its patterns address assets, multi-assets, resources, jobs, schedules, sensors, partitions, checks, retries, and observations.

The basic illustration extracts orders from a CSV file, cleans missing data, and groups results into a summary. A multi-asset example produces bronze, silver, and gold customer outputs in one computation. This structure models related data products but still requires organization-specific sources and transformation logic.

Use it when designing or extending a Dagster pipeline that needs clear data dependencies and operational structure. It can help connect transformation logic to resource configuration, scheduled or event-driven execution, partitioning, and data-quality checks.

Treat the code as illustrative and verify it against the installed Dagster version before execution. Some examples depend on undefined connections and helper functions. The scheduling example also uses a sensor-status constant for a schedule, so its status configuration needs checking. The prompt does not install dependencies, supply credentials, or establish that a pipeline is production-ready.

Inputs include source data, desired assets, dependency relationships, transformation rules, resource requirements, and execution timing. Outputs can include asset definitions, multi-output transformations, configurable resources, selected-asset jobs, schedules, file sensors, partition definitions, checks, and a Definitions entry point.

DatabaseResource wraps a SQLAlchemy engine with a connection string and pool size. S3Resource illustrates a boto3 client and bucket configuration. Credential strings in the examples are placeholders requiring proper environment-specific configuration. Jobs select assets and resources; the daily schedule uses a cron expression, while the file sensor inspects incoming CSV filenames and yields a RunRequest. Its simplified filename selection is a pattern to adapt, not a complete file ingestion policy.

DailyPartitionsDefinition illustrates time-based processing, with the partition key used to select sales for one date. DynamicPartitionsDefinition shows per-product partitions. Source SQL and connection handling need adaptation to the actual database and data model.

The test example uses materialize to execute assets and checks success and emitted materializations. An asset check returns AssetCheckResult based on whether a summary is present and nonempty. Despite its freshness-oriented name, that example does not establish a time-based freshness guarantee.

RetryPolicy and Backoff illustrate retries for transient failures. Import the retry controls with:

from dagster import RetryPolicy, Backoff

The surrounding policy specifies three retries, a thirty-second delay, and exponential backoff. Errors are logged and re-raised. AssetObservation records row counts and null proportions as metadata, complementing checks without itself enforcing a quality threshold.

The prompt is for data engineers organizing Dagster transformations into understandable, testable assets and the operational definitions needed to run and observe them.