A Green Spark Job Isn't Enough: Building a Fail-Closed Pipeline Health Monitor in Python

A Spark pipeline can produce the right files and still be operationally unhealthy. Maybe the Silver stage failed once and recovered. Maybe a process started but never wrote its completion event. Or perhaps a dashboard reads a half-written health report and displays a misleading green status.

For a retail-media analytics pipeline, those distinctions matter: delayed campaign metrics can be mistaken for a drop in performance, while an incomplete Gold refresh can look like a successful reporting cycle.

I recently added a Python-only operational monitor to my public, synthetic-data retail-media analytics pipeline. This article walks through the actual design, its event contract, failure semantics, tests, and limitations. The monitoring implementation, orchestrator, and tests are public. All examples here use synthetic events, not employer or production advertising data.

1. Separate pipeline execution from health interpretation

The demonstration pipeline has two ordered PySpark stages:

Synthetic campaign CSV | v Stage 1: Bronze -> Silver (PySpark) | v Stage 2: Silver -> Gold (PySpark) | v Reporting-ready metrics 

The local orchestrator invokes spark-submit for each stage, stops if a stage exhausts its retries, and appends structured events. The monitor is a separate reader: it does not launch Spark or decide which command to retry. It reconstructs the current health of observed runs from the events already written.

This separation has a practical benefit. A lightweight monitoring job can run without starting a Spark session, and a dashboard can consume its output without parsing free-form executor logs.

2. Use an event contract, not keyword searches

The orchestrator writes one JSON object per line to data/outputs/pipeline_events.jsonl. Every event includes a run identifier, stage number, attempt number, status, and timezone-aware UTC timestamp. Terminal events also carry a return code; failures may include bounded error text.

For example, these are illustrative synthetic events from a stage that fails once and then succeeds:

{"run_id":"demo-001","stage":1,"attempt":1,"status":"started","timestamp_utc":"2026-10-31T12:00:00+00:00"} {"run_id":"demo-001","stage":1,"attempt":1,"status":"failed","returncode":1,"error":"temporary failure","timestamp_utc":"2026-10-31T12:00:03+00:00"} {"run_id":"demo-001","stage":1,"attempt":2,"status":"started","timestamp_utc":"2026-10-31T12:00:05+00:00"} {"run_id":"demo-001","stage":1,"attempt":2,"status":"succeeded","returncode":0,"timestamp_utc":"2026-10-31T12:00:08+00:00"} {"run_id":"demo-001","stage":2,"attempt":1,"status":"started","timestamp_utc":"2026-10-31T12:00:09+00:00"} {"run_id":"demo-001","stage":2,"attempt":1,"status":"succeeded","returncode":0,"timestamp_utc":"2026-10-31T12:00:14+00:00"} 

JSON Lines works well here because events can be appended incrementally and processed line by line. See the JSON Lines format conventions.

Notice the distinction between an attempt and a run. A failed attempt does not necessarily mean a failed run: the next attempt may recover. Conversely, one successful stage does not make a two-stage pipeline successful.

3. Validate before declaring anything healthy

The monitor's summarize_events() function checks the event stream instead of assuming it is trustworthy. Among other rules, it rejects:

• Missing or invalid run IDs, stage numbers, attempts, or statuses.
• Naive timestamps without timezone information and per-run events arriving out of chronological order.
• Duplicate attempt starts, skipped attempt numbers, and retries without a preceding failed attempt.
• Terminal events without an active start, or success events with a nonzero return code.
• Starting stage 2 before stage 1 has succeeded.

A malformed log raises an error rather than being quietly treated as an empty or successful run. That is a deliberate fail-closed policy: a green dashboard should require positive evidence.

The run-level state is calculated from the latest state of each expected stage. A run is succeeded only when every expected stage succeeded. Otherwise, a failed stage takes precedence, followed by a stalled stage, a currently running stage, or an incomplete run. The output also retains attempt counts, retry counts, failed-attempt counts, and the last bounded error for each stage.

4. Detect stalls without confusing them with failures

A process can emit started and then stop producing events. The monitor classifies an unfinished attempt as stalled only when its age exceeds a configurable threshold:

# Excerpted logic from src/pipeline_monitor.py if stage_status == "running" and ( current - last["started"] ).total_seconds() > stale_after_seconds: stage_status = "stalled" 

The default threshold is 600 seconds. A recently started attempt remains running; an old completed attempt remains succeeded, not stalled. The code uses timezone-aware timestamps and permits a fixed now argument, which makes boundary cases deterministic in tests.

There is an important caveat: a stall is inferred from the absence of a completion event, not proven process death. Long legitimate jobs need an appropriate threshold or a heartbeat protocol. A monitoring snapshot can also change if later retry events arrive.

5. Export dashboard JSON atomically

The monitor produces a snapshot with schema_version, generated_at_utc, latest_run_id, run_count, per-status health_counts, and detailed run/stage records. A consumer could load that JSON into a small operational dashboard or Power BI dataflow; the repository does not claim a deployed production dashboard.

Rather than writing directly over the existing report, the implementation writes a temporary file in the destination directory, flushes and syncs it, then replaces the destination with os.replace():

# Simplified illustration of the repository's write pattern with tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", dir=target.parent, delete=False, ) as stream: json.dump(summary, stream, allow_nan=False) stream.flush() os.fsync(stream.fileno()) temp_path = Path(stream.name) os.replace(temp_path, target) 

The real implementation also creates the destination directory and cleans up temporary files in a finally block. Keeping the temporary file in the same directory supports atomic replacement on the same filesystem; see Python's os.replace documentation. If event validation fails, the previous dashboard file is preserved rather than replaced with a misleading partial report.

6. Make the monitor useful in automation

After the sample pipeline has generated events, run:

python -m src.pipeline_monitor \ --events data/outputs/pipeline_events.jsonl \ --output data/outputs/pipeline_health.json \ --expected-stages 2 \ --stale-after-seconds 600 \ --fail-on-unhealthy 

With --fail-on-unhealthy, the exit codes are intentionally meaningful:

Without --fail-on-unhealthy, an otherwise valid snapshot can be written with exit code 0 even if it describes unhealthy runs. That distinction matters when integrating the command into a scheduled check or a CI step. GitHub Actions uses process exit codes to determine success or failure; wiring this monitor into a deployment gate would be a separate integration decision, not something this example claims is already configured.

7. Test state transitions, not only happy paths

The monitor tests use synthetic JSONL files, temporary directories, and a fixed UTC clock. They cover successful two-stage runs, retry-then-success, permanently failed stages, running versus stalled attempts, invalid event ordering, orphan completions, invalid return codes, empty logs, atomic output, and unhealthy CLI exits.

Run the dedicated tests with:

python -m unittest discover -s tests -p 'test_pipeline_monitor.py' -v 

The October 31 implementation commit added the monitor, documentation, and its tests. The corresponding GitHub Actions run passed. That validates this repository's tested behavior; it is not a claim about production reliability or benchmark performance.

8. What this design does not solve

Operational health is not data correctness. A pipeline can finish every stage while still producing bad spend, impressions, or conversion metrics; separate data-quality and KPI checks remain necessary. The monitor also does not provide distributed locks, a durable event broker, multi-worker coordination, heartbeat-based liveness, alert routing, or log retention policies. The JSONL stream can contain command arguments and error text, so production logs would need secret redaction and access controls.

For a production system, I would add run ownership, heartbeats, retention, idempotent event ingestion, structured alerts, and a clear policy for late or out-of-order events before using these snapshots as an operational source of truth.

Takeaway: A trustworthy green status is something you prove from ordered, validated evidence—not something you infer because the last command printed “done.” A small, tested Python monitor can make retries, partial runs, and stalled stages visible without turning every health check into another Spark job.

Code and further reading: Repository · Monitoring guide · JSON Lines · Python atomic replacement