Reliable Data Products: Validation, Delivery, Observability, and Recovery

45 min

Scheduling, Idempotency, and Incremental Runs

Learning objective: design reruns safely; understand full refresh vs incremental processing.

A scheduled run kicks off at 2am and fails halfway through because of a network blip. The on-call engineer manually retriggers it, and the next morning you discover the same batch_id was counted twice in the output table, doubling the total. This is a classic non-idempotent pipeline: rerunning the same input did not overwrite or replace the previous result, it stacked on top of it. Designing safe reruns is a baseline requirement for any scheduled pipeline, not an advanced option.

# Wrong: every run appends, so a rerun creates duplicate rows
result.to_sql("monthly_summary", con=engine, if_exists="append")

# Right: key the write by the run's processing range, delete the same range's old
# result first, then write the new one (idempotent)
def publish_idempotent(result: pd.DataFrame, period: str, engine) -> None:
    with engine.begin() as conn:
        conn.execute(
            "DELETE FROM monthly_summary WHERE period = :period",
            {"period": period}
        )
        result.to_sql("monthly_summary", con=conn, if_exists="append", index=False)

# An incremental run needs to track how far it has already processed (a watermark)
def get_watermark(engine) -> str:
    row = engine.execute("SELECT MAX(period) AS last_period FROM monthly_summary").first()
    return row.last_period if row and row.last_period else "1970-01"

1. 1Check whether the write is idempotent
   Rerunning the same input twice should leave the output table exactly the same, never doubled.
2. 2Choose full refresh or incremental
   Decide based on data volume and whether the source revises history, whether to recompute everything or just the new part.
3. 3Track a watermark
   An incremental pipeline needs to record how far it has already processed, as the starting point for the next run.
4. 4Handle late-arriving data
   Design a mechanism so data that arrives later than expected is still correctly attributed to its period.

Key takeaway: Completing the steps is not enough; the result must be verifiable, reproducible, and explainable.

45 min

Logging, Metrics, Lineage, and Observability

Learning objective: capture run IDs, row counts, data freshness, validation results, failures, lineage.

Your pipeline shows last night’s scheduled run as "succeeded," but your manager complains the monthly report’s numbers are down 90% from normal. You open the log and see a single line, "Job completed successfully", no row counts, no validation results, nothing to tell you which stage went wrong. Logging only a binary success/failure status is like driving with no dashboard: the car is still moving, but you have no idea how much fuel is left or whether the engine is overheating.

import logging, uuid

logger = logging.getLogger("ds301.pipeline")

def run_pipeline(input_path: str) -> None:
    run_id = str(uuid.uuid4())
    logger.info("run_start", extra={"run_id": run_id})

    raw = load_raw(input_path)
    logger.info("ingest_done", extra={"run_id": run_id, "row_count": len(raw)})

    errors = validate_contract(raw)
    logger.info("validation_done", extra={"run_id": run_id, "error_count": len(errors)})

    transformed = fill_missing_values(filter_current_schema(raw, "v2"))
    summary = summarize_by_metric(transformed)
    logger.info(
        "analyze_done",
        extra={"run_id": run_id, "output_row_count": len(summary), "total_value": float(summary["value"].sum())}
    )

    logger.info("run_end", extra={"run_id": run_id, "status": "succeeded"})

• Row count shifts beyond a reasonable threshold compared with the previous run (for example, plus or minus 30%)
• The validation error count goes from zero to nonzero, even if it is currently only warning-level
• A given run_id’s freshness falls behind the scheduled expectation by more than a set threshold
• Lineage records show this run used an unexpected input version or code version

Key takeaway: Completing the steps is not enough; the result must be verifiable, reproducible, and explainable.

45 min

Human Approval, Publishing, and Rollback

Learning objective: place approval before high-impact publish; preserve prior artifacts; define rollback.

Your pipeline is fully automated end to end, publishing straight to an internal dashboard. Last month, an upstream field definition silently changed, and the published monthly metrics undercounted an entire business unit, with no human ever looking at it before it went out, because the flow was fully automatic. By the time a manager spotted the discrepancy in a meeting, the wrong report had already been circulating for two weeks. High-impact data products need a person to confirm the result is reasonable before the "publish to users" step, rather than trusting automation completely.

def publish_with_approval(result: pd.DataFrame, period: str, approved_by: str | None) -> None:
    if approved_by is None:
        # High-impact output: cannot publish without approval; write to a pending-review area
        write_to_pending_review(result, period)
        raise RuntimeError(f"period={period} not yet approved; written to pending review, awaiting human confirmation")

    # Versioned output: keep an identifiable prior version so rollback is possible
    archive_previous_version(period)
    publish_idempotent(result, period, engine)
    log_publish_event(period, approved_by)

def rollback(period: str) -> None:
    restore_previous_version(period)
    log_rollback_event(period)

• After publishing, you discover the upstream data had already quietly degraded before publish, but the contract check did not catch it
• The approver changes their mind afterward and decides this result should not have been published
• A downstream user reports that the numbers clearly do not match their own records
• You discover this publish used the wrong code version or configuration

Key takeaway: Completing the steps is not enough; the result must be verifiable, reproducible, and explainable.

45 min

Testing, Failure Recovery, and Runbooks

Learning objective: use unit/data/integration smoke tests; define retries, dead-letter/manual intervention, runbook.

At 3am, a pipeline fails because an upstream API times out temporarily. The schedule is configured to retry indefinitely on failure, so it retries and fails all night long, and by morning you find the system has run out of resources, while the real problem (the upstream API recovered hours ago, but the code has a separate, permanent bug) was never actually addressed. A reliable workflow needs three things at once: tests that catch known errors before they happen, a disciplined retry strategy that distinguishes transient from permanent failures, and a runbook so whoever is on call knows what to do when something fails.

import time

class TransientError(Exception): ...
class PermanentError(Exception): ...

def call_upstream_with_retry(fetch_fn, max_attempts: int = 3) -> dict:
    for attempt in range(1, max_attempts + 1):
        try:
            return fetch_fn()
        except TransientError as e:
            if attempt == max_attempts:
                raise
            time.sleep(2 ** attempt)  # exponential backoff so we don't hammer the upstream service
        except PermanentError:
            # Permanent error (e.g. malformed data): retrying will never succeed; log it and hand off to a human
            log_for_manual_intervention(fetch_fn.__name__)
            raise

1. 1Distinguish error types
   Split errors into transient (worth retrying) and permanent (retrying is pointless).
2. 2Set a retry cap
   Avoid unlimited retries that waste resources or stall the entire schedule.
3. 3Define when a human steps in
   A permanent error or exceeding the retry cap should notify a person, not keep retrying.
4. 4Write a runbook
   Document common failure scenarios, how to diagnose them, response steps, and who to escalate to.

Key takeaway: Completing the steps is not enough; the result must be verifiable, reproducible, and explainable.

60 min

Independent Lab: Reliable Analytics Workflow

Learning objective: independently deliver an executable prototype, architecture canvas, tests, observability plan, runbook, and ownership model.

• Go to the DS 301 Practice page
• PreviewDownload the practice analytics package with challenges (synthetic data)
• Download the Independent Lab starter notebook / template
• PreviewDownload the data dictionary (Traditional Chinese)
• PreviewDownload the data dictionary (English)

1. 1Input Contract
   Write down this workflow’s input data contract and the assumptions it allows.
2. 2Executable Prototype
   Deliver a runnable pipeline covering the five ingest→validate→transform→analyze→publish stages.
3. 3Tests
   Include at least one unit test and one data validation check.
4. 4Observability
   Log run IDs, row counts, validation results, and lineage.
5. 5Human Approval & Rollback
   Describe the human approval point before publish and how to roll back to a prior version.
6. 6Runbook
   Write down how to diagnose, respond to, and escalate at least two common failure scenarios.
7. 7Ownership Model
   Explicitly state who owns this data product and what a handoff needs to know.

• Only testing the happy path, without verifying the pipeline actually fails when data violates the contract
• Observability only logs success or failure, without row counts or validation results
• The runbook is too vague (for example, just "contact an engineer"), with no concrete diagnosis steps or escalation target
• The ownership model lists only a name, without explaining what background knowledge a handoff actually needs

Key takeaway: Completing the steps is not enough; the result must be verifiable, reproducible, and explainable.