Day 19. Schedules, sensors, retries and alerts

⏱ 4h · learn 65m · lab 135m · ship 25m · review 15m🏷 hospitality, retailBefore: Day 18

By the end of today you can

Schedule partitioned jobs with the right partition for each run
Write file and email sensors with cursors and run keys
Configure retries with backoff and know what not to retry
Send failure alerts that say what to do

1. Why this matters

The 06:00 timeline from Day 1 becomes code today: the hotel file arrives "around 04:00, sometimes late"; emails arrive from 06:00; reports are due at 08:00. Schedules handle the predictable, sensors handle the unpredictable, retries handle the flaky, and alerts handle the rest.

2. Schedules

hotel_job = dg.define_asset_job("hotel_daily", selection=dg.AssetSelection.groups("hotel"), partitions_def=daily)

# runs at 06:15 Toronto time for *yesterday's* partition
hotel_schedule = dg.build_schedule_from_partitioned_job(hotel_job, hour_of_day=6, minute_of_hour=15)

build_schedule_from_partitioned_job targets the last complete partition, so the 06:15 run on 15 Sept processes 14 Sept. That's the two-clocks idea from Day 15, built in.

3. Sensors

A sensor is a small function Dagster's daemon calls every ~30 s. It looks at the world and returns run requests.

@dg.sensor(job=hotel_job, minimum_interval_seconds=120)
def pms_file_sensor(context: dg.SensorEvaluationContext, sftp: SftpSource):
    seen = set(json.loads(context.cursor or "[]"))
    ready = [f for f in sftp.list_ready() if f not in seen]       # .partial files are not "ready"
    for f in ready:
        bdate = datetime.strptime(f.split("_")[-1][:8], "%Y%m%d").date().isoformat()
        yield dg.RunRequest(run_key=f, partition_key=bdate, tags={"source_file": f})
    context.update_cursor(json.dumps(sorted(seen | set(ready))[-500:]))

run_key makes the sensor idempotent: Dagster never launches two runs with the same key, even if the sensor sees the file twice. The cursor stores state between evaluations.

For email, the same pattern: list messages received after the cursor's watermark (Day 4), request one run per business date.

4. Retries

@dg.asset(retry_policy=dg.RetryPolicy(max_retries=3, delay=30, backoff=dg.Backoff.EXPONENTIAL, jitter=dg.Jitter.PLUS_MINUS))
def raw_ads(...): ...

Retry transient failures (timeouts, 5xx, locked files). Don't retry deterministic ones (schema violation, bad credentials, a bug): it just delays the alert and can hammer a source. Raise a specific exception and catch transient ones in your client (Day 6) so the asset-level retry is a last line.

Maths and principle

Principle: retries raise success probability fast when failures are independent, and not at all when they're correlated.

Equation: with independent per-attempt failure probability $p$, success within $n$ attempts is $P = 1 - p^n$. For $p = 0.1$, three attempts give $0.999$. If the failure is deterministic, $p = 1$ and $P = 1 - 1^n = 0$ for every $n$.

Analogy: repeated coin flips vs a two-headed coin. Where it breaks: a source down for an hour makes failures correlated over time. Backoff spreads attempts so some land after recovery.

Try it: $p=0.2$, how many attempts for ≥ 99.9% success? (n = 5: 1 − 0.2⁵ = 0.99968.)

5. Alerts that help

@dg.run_failure_sensor(monitored_jobs=[hotel_job])
def notify_on_failure(context: dg.RunFailureSensorContext):
    err = context.failure_event.message
    send_teams_or_slack(f"❌ {context.dagster_run.job_name} failed for partition "
                        f"{context.dagster_run.tags.get('dagster/partition')}: {err[:300]}\n"
                        f"Runbook: docs/runbooks/hotel.md#failures")

Good alerts name the job, the partition, the first line of the error, and the runbook link. Integrations exist for Slack (dagster-slack), Microsoft Teams (dagster-msteams), email, and PagerDuty.

The 5-year lens

Design for the late file. The schedule at 06:15 runs with whatever arrived; the sensor picks up the late file at 07:40 and reprocesses that partition; the report job at 07:55 runs if the data is fresh, and otherwise sends the report with a clear "Hotel H3 data not yet received, figures exclude H3" banner instead of silently showing lower occupancy. That banner decision is a business conversation, and having it before the incident is what seniority looks like.

Lab

Build hotel_job and its 06:15 schedule. Turn it on in the UI; use Test schedule to see which partition it targets.
Write pms_file_sensor with run keys and cursor. Drop a new file into the local SFTP folder; watch a run start. Drop the same file again; confirm no new run. Rename a .partial; confirm it triggers.
Add a retry policy to the SFTP and API assets. Make the fake API fail 50% of the time and watch retries in the run log.
Add a run-failure sensor that writes alerts to data/alerts.log (and to a Slack or Teams webhook if you have a free workspace).
Write docs/runbooks/hotel.md: each alert, what it means, the first three things to check.

Ship

feat(dagster): schedule, idempotent file sensor, retries, failure alerts, runbook.