Automating Your Pipeline: Daily Schedules and Historical Backfills

Your pipeline is deployed and can materialize assets on demand. But for production use, you need automation—daily schedules that run without intervention and backfills to process historical data. This chapter covers making your pipeline fully autonomous.

Schedule Strategy

AIS data follows a predictable pattern:

• DMA: Publishes daily files around 03:00 UTC
• Data availability: Yesterday’s data available today

Our daily schedule will:

• Run at 06:00 UTC (safe margin after data publishes)
• Process yesterday’s partition (D-1 pattern)
• Materialize all assets (bronze → silver → validation → gold)

Creating Schedules

Define schedules in your pipeline code:

# src/pipeline/schedules.py
from dagster import (
    schedule,
    RunRequest,
    SkipReason,
    DefaultScheduleStatus,
)
from datetime import datetime, timedelta

@schedule(
    cron_schedule="0 6 * * *",  # 06:00 UTC every day
    execution_timezone="UTC",
    default_status=DefaultScheduleStatus.RUNNING,
)
def daily_dma_ingestion(context):
    """Daily DMA AIS data ingestion schedule.

    Runs at 06:00 UTC to process yesterday's data (D-1 pattern).
    DMA publishes daily files around 03:00 UTC, so 06:00 gives safe margin.
    """
    scheduled_date = context.scheduled_execution_time
    yesterday = scheduled_date - timedelta(days=1)
    partition_key = yesterday.strftime("%Y-%m-%d")

    return RunRequest(
        run_key=f"dma_{partition_key}",  # Unique, idempotent
        partition_key=partition_key,
        tags={
            "schedule": "daily_dma_ingestion",
            "source": "dma",
            "partition_date": partition_key,
        },
        asset_selection=[
            "dma_ais_bronze",
            "dma_ais_silver",
            "dma_silver_validation",
            "vessel_tracks",
            "port_density",
            "daily_stats",
        ],
    )

Conditional Schedules

Skip execution based on conditions:

@schedule(
    cron_schedule="0 6 * * 1-5",  # Monday-Friday only
    execution_timezone="UTC",
    default_status=DefaultScheduleStatus.STOPPED,
)
def weekday_only_ingestion(context):
    """DMA ingestion on weekdays only."""
    scheduled_date = context.scheduled_execution_time

    if scheduled_date.weekday() >= 5:
        return SkipReason("Skipping weekend processing")

    yesterday = scheduled_date - timedelta(days=1)
    partition_key = yesterday.strftime("%Y-%m-%d")

    return RunRequest(
        run_key=f"dma_weekday_{partition_key}",
        partition_key=partition_key,
        asset_selection=["dma_ais_bronze", "dma_ais_silver"],
    )

Smart Schedules with Data Availability Checks

def check_dma_data_available(partition_date: str) -> bool:
    """Check if DMA data is available for partition."""
    import requests

    url = f"https://web.ais.dk/aisdata/aisdk-{partition_date}.zip"

    try:
        response = requests.head(url, timeout=10)
        return response.status_code == 200
    except:
        return False

@schedule(
    cron_schedule="0 6 * * *",
    execution_timezone="UTC",
    default_status=DefaultScheduleStatus.STOPPED,
)
def smart_dma_ingestion(context):
    """Smart schedule that checks data availability first."""
    yesterday = context.scheduled_execution_time - timedelta(days=1)
    partition_key = yesterday.strftime("%Y-%m-%d")

    if not check_dma_data_available(partition_key):
        return SkipReason(f"DMA data not yet available for {partition_key}")

    return RunRequest(
        run_key=f"dma_smart_{partition_key}",
        partition_key=partition_key,
        asset_selection=["dma_ais_bronze", "dma_ais_silver"],
    )

Registering Schedules

Add schedules to your Dagster definitions:

# src/pipeline/repository.py
from dagster import Definitions
from pipeline.assets import all_assets
from pipeline.schedules import daily_dma_ingestion

defs = Definitions(
    assets=all_assets,
    resources={
        "minio_client": minio_resource,
        "duckdb_conn": duckdb_resource,
    },
    schedules=[
        daily_dma_ingestion,
    ],
)

Activating Schedules

Via Dagit UI

1. Open Dagit → Automation tab
2. Find daily_dma_ingestion
3. Toggle switch to ON
4. Check Next tick shows the correct time

Via CLI

# Start a schedule
dagster schedule start daily_dma_ingestion -m pipeline.repository

# Stop a schedule
dagster schedule stop daily_dma_ingestion -m pipeline.repository

# List all schedules
dagster schedule list -m pipeline.repository

# Preview next ticks
dagster schedule preview daily_dma_ingestion -m pipeline.repository

Verify in Kubernetes

kubectl logs -n ais-pipeline deployment/dagster-daemon --tail=100 | grep schedule

# Expected output:
# "Checking for new runs for schedule daily_dma_ingestion"
# "Launching run for schedule daily_dma_ingestion with partition 2024-01-15"

Backfills: Processing Historical Data

Want to process January 2024? Instead of triggering 31 manual runs, use backfills.

Via Dagit UI

1. Open Assets tab
2. Select assets to backfill
3. Click Materialize → Launch a backfill
4. Select partition range (Start: 2024-01-01, End: 2024-01-31)
5. Click Launch backfill

Dagit shows:

• Progress: 0/31 completed
• Visual progress bar
• Duration elapsed

Via CLI

# Backfill a date range
dagster asset materialize \
  --select dma_ais_bronze dma_ais_silver dma_silver_validation \
  --partition-range 2024-01-01 2024-01-31 \
  -m pipeline.repository

# Backfill single partition
dagster asset materialize \
  --select dma_ais_bronze+ \
  --partition 2024-01-15 \
  -m pipeline.repository

The + suffix means “this asset and all downstream dependencies.”

Backfill Strategies

Sequential (default):

• One partition at a time
• Safer for resource-constrained systems
• Predictable memory usage

Parallel:

• Multiple partitions simultaneously
• Faster but uses more resources

Configure in Dagster daemon:

# k8s/dagster/values.yaml
dagsterDaemon:
  env:
    DAGSTER_BACKFILL_PARALLELISM: "4"  # Process 4 partitions at once

Chunked:

• Process in batches (e.g., 7 days at a time)
• Balances speed and resource usage

def chunked_backfill(start_date, end_date, chunk_days=7):
    """Backfill in weekly chunks."""
    current = datetime.strptime(start_date, "%Y-%m-%d")
    end = datetime.strptime(end_date, "%Y-%m-%d")

    while current <= end:
        chunk_end = min(current + timedelta(days=chunk_days - 1), end)
        print(f"Backfilling {current.date()} to {chunk_end.date()}")

        # Run backfill for this chunk
        # dagster asset materialize --partition-range ...

        current = chunk_end + timedelta(days=1)

Idempotent Processing

Your pipeline should be idempotent—running the same partition twice produces the same result.

Why Idempotency Matters

• Retry failures: Re-run failed partitions without duplicates
• Late data: Reprocess when updated data arrives
• Schema evolution: Reprocess old data with new logic

Built-in Idempotency

Bronze Layer:

def download_day(self, target_date, skip_if_exists=True):
    object_path = f"bronze/dma/aisdk-{target_date}.csv.gz"

    if skip_if_exists:
        try:
            self.minio_client.stat_object("ais-lake", object_path)
            return object_path  # Already exists, skip
        except:
            pass

    # Download...

Silver Layer:

• DuckDB COPY ... TO overwrites existing Parquet files
• No duplicate data even if run twice

Gold Layer:

• Partitioned output overwrites same partition
• Aggregations are deterministic

Testing Idempotency

# Run partition twice
dagster asset materialize --select dma_ais_silver --partition 2024-01-15
dagster asset materialize --select dma_ais_silver --partition 2024-01-15

# Check MinIO: Same file, not duplicate
mc ls ais-minio/ais-lake/silver/ais/source=dma/dt=2024-01-15/

Handling Late-Arriving Data

When data is republished with corrections:

Manual Reprocessing

dagster asset materialize --select dma_ais_bronze+ --partition 2024-01-15

Sensor for Updated Data

from dagster import sensor, RunRequest

@sensor(minimum_interval_seconds=3600)
def dma_update_sensor(context):
    """Detect when DMA republishes files."""
    updated_partitions = []

    # Check last 30 days for file updates
    for days_ago in range(30):
        date = datetime.now() - timedelta(days=days_ago)
        partition_key = date.strftime("%Y-%m-%d")

        current_etag = get_file_etag(f"bronze/dma/aisdk-{partition_key}.csv.gz")
        last_etag = context.cursor

        if current_etag != last_etag:
            updated_partitions.append(partition_key)

    if updated_partitions:
        return [
            RunRequest(
                run_key=f"dma_update_{p}",
                partition_key=p,
                asset_selection=["dma_ais_bronze", "dma_ais_silver"],
            )
            for p in updated_partitions
        ]

Monitoring Scheduled Runs

Dagit UI

Runs Tab:

• Filter: tags:schedule=daily_dma_ingestion
• Check failure rate and duration trends

Automation Tab:

• Schedule tick history
• Next scheduled execution
• Skipped ticks (indicates daemon issues)

Prometheus Alerts

- alert: AISScheduledRunFailed
  expr: increase(ais_asset_failures_total{asset_name="dma_ais_silver"}[1h]) > 0
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: "Scheduled AIS run failed"

- alert: AISScheduleMissed
  expr: (time() - dagster_schedule_last_tick_timestamp{schedule_name="daily_dma_ingestion"}) > 86400
  for: 1h
  labels:
    severity: critical
  annotations:
    summary: "AIS schedule not running"
    description: "Schedule hasn't run in 24 hours"

Best Practices

Partition Naming

Use ISO 8601 format: YYYY-MM-DD

• Sorts correctly
• Unambiguous
• SQL compatible

Schedule Timing

• After data availability: Schedule 3+ hours after source publishes
• Before business hours: Complete before users need data
• Stagger schedules: Don’t run everything at once

Error Handling

• Transient failures: Retry automatically (schedule runs tomorrow)
• Persistent failures: Alert, investigate, fix, re-run
• Partial failures: Asset-level isolation

Data Retention

• Bronze: Delete after 7 days (temporary staging)
• Silver: Keep indefinitely (source of truth)
• Gold: Keep indefinitely (or archive yearly)

What You’ve Built

Your pipeline now runs autonomously:

• Daily schedules: Automatic D-1 ingestion at 06:00 UTC
• Backfills: Process historical data in bulk
• Idempotency: Safe reprocessing of partitions
• Late data handling: Reprocess updated partitions
• Monitoring: Track scheduled runs and failures

What’s Next

With automation in place, Chapter 12 covers day-to-day operations—health checks, troubleshooting, and maintenance tasks.

---

Building automated data pipelines? We help organizations design and implement hands-free data infrastructure. Get in touch to discuss your project.