Schedule Pipelines with Dagster

By Khimananda Oli | Last reviewed: August 2026

Data workflows fail silently when triggered by fragile cron jobs or disconnected scripts. If you need to schedule pipelines with Dagster, you are moving beyond simple time-based triggers to a system that understands data dependencies, state, and observability. This guide covers the practical implementation of Schedules and Sensors, ensuring your automation is as reliable as the infrastructure it runs on, a critical step for teams adopting modern MLOps practices.

How do you configure time-based schedules in Dagster?

When you schedule pipelines with Dagster using fixed intervals, you use the ScheduleDefinition class. Unlike a Linux crontab entry buried in /etc/cron.d/, a Dagster schedule is a Python object defined alongside your asset code. This co-location means your scheduling logic is version-controlled, testable, and visible in the Dagster UI.

Defining a basic cron schedule

The most common pattern is triggering an asset job at a specific cadence. In 2026, Dagster’s API favors explicit job targeting over legacy pipeline references. Here is a production-grade configuration for an hourly ETL job:

from dagster import ScheduleDefinition, define_asset_job, AssetSelection

# Define the job targeting specific assets
hourly_etl_job = define_asset_job(
    name="hourly_customer_etl",
    selection=AssetSelection.groups("customer_analytics"),
)

# Bind the schedule to the job
hourly_etl_schedule = ScheduleDefinition(
    name="hourly_customer_etl_schedule",
    job=hourly_etl_job,
    cron_schedule="0 * * * *",  # Top of every hour
    execution_timezone="Asia/Kathmandu",  # Critical for Nepal-based teams
    default_status=DefaultScheduleStatus.STOPPED,  # Safe deployment practice
)

A common mistake I see in audits is omitting execution_timezone. Without it, Dagster defaults to UTC. For teams operating in NPT (UTC+5:45), this causes confusion when business stakeholders expect reports at 8:00 AM local time but the pipeline runs at 2:15 AM. Always set the timezone explicitly to match your business domain.

Parameterizing scheduled runs

Static cron expressions only get you so far. You often need to pass dynamic configuration based on the scheduled time. Use the configured method or a config function to inject partition keys or runtime parameters:

def hourly_config(context):
    """Generate run config dynamically at schedule evaluation time."""
    return {
        "ops": {
            "extract_data": {
                "config": {
                    "window_start": context.scheduled_execution_time.strftime("%Y-%m-%dT%H:00:00"),
                    "source_env": "production"
                }
            }
        }
    }

parameterized_schedule = ScheduleDefinition(
    name="hourly_etl_with_config",
    job=hourly_etl_job,
    cron_schedule="0 * * * *",
    run_config_fn=hourly_config,
    execution_timezone="Asia/Kathmandu",
)

This approach eliminates hardcoded values and makes your schedule adaptable. When debugging failed runs, the exact configuration used is captured in the run metadata, which is invaluable for post-incident analysis and aligns with structured logging best practices.

When should you use Dagster sensors instead of schedules?

Schedules assume time is the trigger. Sensors assume state is the trigger. Use a SensorDefinition when you need to schedule pipelines with Dagster based on external events: new files in S3, database row counts exceeding a threshold, or API webhook receipts.

Implementing a file-based sensor

Sensors maintain a cursor between evaluations. This prevents reprocessing the same event. Below is a robust sensor that checks for new CSV files in an S3-compatible bucket (common in hybrid Nepal cloud setups using MinIO or Cloudflare R2):

from dagster import sensor, RunRequest, SkipReason
import boto3

@sensor(job=process_upload_job, minimum_interval_seconds=60)
def s3_ingest_sensor(context):
    s3 = boto3.client('s3')
    bucket = "raw-data-ingest"
    
    # Retrieve last processed key from persistent cursor
    last_key = context.cursor or ""
    
    response = s3.list_objects_v2(Bucket=bucket, Prefix="uploads/", StartAfter=last_key)
    contents = response.get("Contents", [])
    
    if not contents:
        return SkipReason(f"No new files after {last_key}")
    
    run_requests = []
    latest_key = last_key
    
    for obj in contents:
        if obj["Key"].endswith(".csv"):
            run_requests.append(
                RunRequest(
                    run_key=obj["Key"],  # Idempotency key prevents duplicate runs
                    run_config={
                        "ops": {"load_csv": {"config": {"s3_key": obj["Key"]}}}
                    },
                    tags={"source_file": obj["Key"]}
                )
            )
            latest_key = max(latest_key, obj["Key"])
    
    # Update cursor only after successful request generation
    context.update_cursor(latest_key)
    return run_requests

Note the use of run_key. This is non-negotiable for production sensors. If the sensor evaluates twice before the cursor updates, Dagster uses the run_key to deduplicate requests. Without it, you risk double-processing data, which breaks downstream aggregations and violates idempotency principles essential for reliable embedding pipelines.

How do Dagster schedules compare to Airflow and Cron?

Choosing the right tool requires honest comparison. Many teams in Nepal still rely on VPS-level cron or are migrating from Airflow. Understanding the trade-offs helps justify the operational shift when you schedule pipelines with Dagster.

Cron is fine for single-server maintenance tasks. Airflow excels at complex DAG orchestration with heavy dependency chains. Dagster wins when your scheduling logic is tightly coupled to data assets and you need software-engineering rigor applied to automation. The ability to unit-test a schedule definition locally before deploying to production is a massive reliability win that neither cron nor Airflow offers ergonomically.

What are the best practices for testing and monitoring Dagster automation?

Untested schedules are ticking time bombs. When you schedule pipelines with Dagster, treat the schedule and sensor definitions as first-class code artifacts.

Unit testing schedules

Dagster provides utilities to evaluate schedules without triggering actual runs. This catches timezone errors, config generation bugs, and partition mismatches before they hit production:

from dagster import validate_run_config

def test_hourly_schedule_config():
    """Verify schedule generates valid run configuration."""
    context = build_schedule_context(
        scheduled_execution_time=datetime(2026, 8, 15, 10, 0, 0),
        instance=DagsterInstance.ephemeral()
    )
    
    result = hourly_etl_schedule.evaluate_tick(context)
    
    assert len(result.run_requests) == 1
    run_config = result.run_requests[0].run_config
    validate_run_config(hourly_etl_job, run_config)
    assert "2026-08-15T10:00:00" in str(run_config)

Monitoring scheduler health

The Dagster daemon process must be running for schedules and sensors to execute. In Kubernetes deployments, this is typically a separate Deployment or part of the Helm chart. Monitor these signals:

• Daemon heartbeat: Alert if no heartbeat for >2 minutes
• Sensor tick duration: P95 latency indicates external API degradation
• Skip reason rate: Unexpected spikes may indicate source system issues
• Evaluation failures: Track as errors; these mean missed runs

Integrate these metrics into your existing stack. If you are already running Prometheus and Grafana, export Dagster’s internal metrics via the OpenTelemetry integration or StatsD exporter. Set up alerts for consecutive evaluation failures — three failures in a row usually means a broken config or credential expiration, not transient network noise.

Operationalizing Your Dagster Automation

To reliably schedule pipelines with Dagster, start with ScheduleDefinition for predictable workloads and graduate to SensorDefinition only when external state truly drives execution. Always set explicit timezones, implement idempotency keys, and write unit tests for your schedule logic before deploying. Monitor the daemon process as critically as any database replica.

If your team is struggling with silent cron failures or needs to migrate fragile Airflow DAGs to a more testable framework, reach out to discuss your automation architecture. Getting scheduling right is foundational to trustworthy data systems.