Data Pipeline Monitoring and Observability Questions

Direct answer

Instrument a streaming worker with three Prometheus client types matched to what they measure: a Counter for a monotonically-increasing count (processed events), a Histogram for a distribution you want percentiles on (processing latency), and a Gauge for a value that goes up and down (current in-flight tasks), each labeled with the pipeline name and stage so they can be filtered and aggregated consistently with the platform's naming convention.

Structured elaboration

• Counter: only ever increases (or resets to zero on restart), appropriate for "total events processed" since you never want to decrement it, PromQL's rate() function is built to handle counter resets correctly.
• Histogram: buckets observations into ranges and lets you compute percentiles (p50, p95, p99) after the fact via PromQL, the right choice for latency because you almost always care about the tail (p99), not just the average.
• Gauge: can go up or down freely, correct for "in-flight tasks" since that number naturally increases when work starts and decreases when it completes.
• Labels: attach pipeline_name and stage to every metric so a single Grafana dashboard can filter or group by either dimension across all instrumented workers, consistent with the shared naming/tagging convention (avoid attaching a per-task or per-record id as a label, which would create unbounded cardinality).

Worked example

from prometheus_client import Counter, Histogram, Gauge

processed_events = Counter(
    'pipeline_events_processed_total',
    'Total events processed by this worker',
    ['pipeline_name', 'stage']
)
processing_latency = Histogram(
    'pipeline_processing_latency_ms',
    'Event processing latency in milliseconds',
    ['pipeline_name', 'stage'],
    buckets=[10, 25, 50, 100, 250, 500, 1000, 2500, 5000]
)
in_flight_tasks = Gauge(
    'pipeline_in_flight_tasks',
    'Number of tasks currently being processed',
    ['pipeline_name', 'stage']
)

def process_event(event, pipeline_name, stage):
    in_flight_tasks.labels(pipeline_name, stage).inc()
    start = time.monotonic()
    try:
        result = do_processing(event)
        processed_events.labels(pipeline_name, stage).inc()
        return result
    finally:
        elapsed_ms = (time.monotonic() - start) * 1000
        processing_latency.labels(pipeline_name, stage).observe(elapsed_ms)
        in_flight_tasks.labels(pipeline_name, stage).dec()

I traced through the increment/decrement logic by hand: in_flight_tasks increments before processing starts and decrements in a finally block, so it correctly decrements even if do_processing raises an exception, avoiding a metric that only ever climbs due to unhandled errors leaking a task count that never gets released. The Histogram's observe() call also happens in the finally block, so latency is recorded for both successful and failed processing attempts, which is the behavior you want since a slow failure is still useful diagnostic information.

Trade-offs and pitfalls

Explicit histogram bucket boundaries ([10, 25, 50, ...] milliseconds) should be chosen based on the actual expected latency distribution for this specific worker, default bucket boundaries in most client libraries are tuned for web-request latencies (often centered around 100ms-1s) and can produce a histogram with almost all observations landing in one or two buckets if your pipeline's real latencies are much faster or slower, making percentile estimates from the histogram imprecise. The main cardinality risk to avoid, as with any Prometheus instrumentation, is adding a label with unbounded values, a raw event id or task id as a label would multiply the number of distinct time series unboundedly and is exactly the anti-pattern the shared tagging convention exists to prevent; per-event detail belongs in structured logs, not as a metric label.

Direct answer

An example: a nullable field silently becomes non-nullable, an enum gains a new value nobody downstream expected, or a timestamp format changes from ISO 8601 to Unix epoch. None of these raise an error on their own, a stricter nullability constraint doesn't reject anything if nulls simply stop appearing, and a new enum value or reformatted timestamp parses "successfully" while meaning something different than before, so a downstream consumer silently mis-processes the field.

Structured elaboration

Take the timestamp-format change specifically: a field that used to arrive as "2026-07-22T14:30:00Z" starts arriving as 1753194600 (Unix epoch seconds). A downstream consumer with a loose string-based date parser might not error, it might parse the numeric string as SOMETHING, producing a date wildly in the past or throwing off a date-range filter silently rather than obviously.

Detection: an automated schema/format monitor that samples a percentage of incoming values for a known field and validates them against the expected format (a regex or type check for the timestamp shape), alerting when the pass rate for that format check drops, rather than relying on the field's declared type alone (which a loosely-typed system may not enforce).

Default reaction when detected: the safest default is QUARANTINE, route the affected batch to a dead-letter location and hold it out of the main pipeline, rather than either blocking the whole pipeline (which stops unrelated, unaffected data too) or silently coercing and continuing (which is what let the problem go undetected in the first place). Quarantine buys time to confirm the change is real and intentional (in which case you update the parser and replay) or a genuine upstream bug (in which case you escalate to the source team) before any bad data reaches consumers.

Worked example

Concretely, an automated check samples 1% of incoming event_timestamp values each hour and validates each against the ISO 8601 regex pattern. Normally this passes at ~100%. One hour, the pass rate drops to 40%, and the check fires a WARNING (not yet a hard block, since a small fraction of malformed values might be a pre-existing, tolerable noise level). Fifteen minutes later the pass rate is 0%, crossing a hard threshold, and the pipeline automatically quarantines the current batch while alerting the owning team with the specific sample values that failed the regex, which is what actually lets someone quickly recognize "oh, this looks like epoch seconds now" instead of starting from a bare "format check failed" message.

Trade-offs and pitfalls

Sampling instead of validating every record keeps the check cheap at high throughput, but it introduces a detection lag proportional to the sample rate, a low-frequency but real format regression could hide in the unsampled majority for a while; the mitigation is to sample a higher percentage for fields flagged as business-critical. The other pitfall is defaulting to "coerce and continue" for convenience (silently trying to parse whatever comes in and moving on), which is exactly the behavior that lets subtle schema changes propagate undetected in the first place, since it optimizes for the pipeline never visibly failing rather than for catching the change.

Direct answer

Computing data completeness across many distributed, eventually-consistent partitions, where late-arriving data and duplicate writes are both possible, needs a completeness definition anchored to a fixed EXPECTED count computed once per logical unit of data (not recomputed differently each time you check), combined with a way to count actual arrivals that's immune to double-counting a record that gets reprocessed.

Structured elaboration

• Handling late-arriving data: define completeness relative to a fixed, agreed retention/lateness window (for example, "completeness as of 24 hours after the partition's nominal time"), rather than checking completeness at an arbitrary moment and treating a still-arriving partition as permanently incomplete; report completeness as PROVISIONAL before that window closes and FINAL after.
• Avoiding double-counting on reprocessing: count DISTINCT record identifiers (not raw row counts) toward the completeness numerator, so if a record is written, then reprocessed and rewritten due to a retry, it's counted once, not twice; this requires records to carry a stable, idempotency-safe identifier that survives reprocessing.
• Handling retention windows: for a partition whose retention window has closed, the completeness figure becomes fixed/historical; a new record arriving after that point (a very late arrival) either gets excluded from the completeness calculation entirely (documented as an accepted limitation) or triggers an explicit RECOMPUTATION of that historical completeness figure, a deliberate choice, not an accident, since silently updating historical figures without flagging the change can itself cause confusion for anyone who already reported on the earlier number.
• Surfacing the SLI without double-counting: completeness = COUNT(DISTINCT record_id WHERE arrival_time <= partition_time + lateness_window) / expected_count, computed once the lateness window closes, reported as the authoritative figure, with a clearly-labeled provisional figure available earlier for teams that want an early read.

Worked example

Concretely: partition 2026-07-22 has an agreed 6-hour lateness window (data must fully settle by 06:00 the next day). At 00:00 (mid-window), a provisional completeness check shows 94% of the expected count has arrived, explicitly labeled "provisional, window closes at 06:00." By 06:00, the distinct-record-id count reaches 99.2% of expected, and this becomes the FINAL, authoritative completeness figure for that partition, unaffected by whether any individual record happened to be retried and rewritten during that window, since the distinct-id counting logic already deduplicates any such retries.

Trade-offs and pitfalls

Distinguishing PROVISIONAL from FINAL completeness explicitly, rather than reporting a single ambiguous number at all times, is what avoids the two failure modes of either declaring a still-settling partition falsely complete too early, or perpetually treating a genuinely complete partition as suspiciously incomplete because you never define a point at which "complete enough" is declared final. The pitfall in the distinct-id deduplication approach is that it depends entirely on every record actually carrying a stable, correct identifier, if the identifier itself isn't reliably unique (a poorly-designed key that occasionally collides across genuinely different records), the completeness calculation inherits that flaw silently, so validating identifier uniqueness is a prerequisite for trusting this completeness SLI at all.

Direct answer

A service level indicator (SLI) is the actual measured metric, for example "the fraction of daily reports available by 07:00." A service level objective (SLO) is the internal target for that indicator, for example "99% of days over a rolling 30-day window." A service level agreement (SLA) is the external, often contractual, commitment, usually set looser than the SLO to leave error-budget margin, for example "reports available by 08:00 on 95% of days, with a defined remedy if missed."

Structured elaboration

For a daily reporting pipeline that must be ready by 07:00 with completeness at or above 99.5%:

• SLI: two indicators, "minutes past 07:00 that the report was actually available" and "percentage of expected rows present at publish time."
• SLO: "99% of days, the report is available by 07:00 with completeness ≥ 99.5%," measured over a rolling 30-day window so a single bad day does not permanently break the target.
• SLA: what you promise the business, typically looser, for example "report available by 08:00 on at least 95% of days," with an escalation or credit if breached repeatedly.

Worked example

Concretely, you would instrument a job-completion timestamp event and a row-count-vs-expected-baseline comparison at publish time. Each day produces one data point: (published_at, completeness_pct). Over a rolling 30-day window, count the days where published_at is at or before 07:00 AND completeness_pct is at least 99.5. If 29 of the last 30 days met both criteria, compliance is 29/30, about 96.7%, which is BELOW the 99% target: a single bad day in a 30-day window is already an SLO violation worth investigating, since 99% of 30 days rounds up to needing all 30 to pass. That is a useful, slightly counterintuitive fact about a 99% target measured over a short window: it has essentially zero tolerance for even one bad day at that window length. Widening the look-back to a rolling 90-day window changes the picture: 89 good days out of 90 is 89/90, about 98.9%, still below 99% but closer, while 90/90 is the only way to clear 99% outright at that window length too, illustrating why teams often pick a window long enough (or a target loose enough) that the SLO can absorb an occasional bad day without every single miss becoming a declared violation.

Trade-offs and pitfalls

The common mistake is treating the SLO and SLA as the same number. If your internal SLO and external SLA are identical, you have zero error budget: any single miss is simultaneously an internal target failure and a customer-facing breach, which pushes teams toward reactive firefighting instead of using the budget deliberately (for example, deferring a risky migration until the budget has recovered). Another pitfall is defining the SLI too loosely, "the pipeline ran," instead of tying it to what the consumer actually cares about, "the data was both on time and complete," since a pipeline can run successfully and still produce an incomplete or stale report.

Direct answer

Instrumenting a Spark ETL job with OpenTelemetry means wrapping each meaningful stage of the job (reading from a source, a shuffle-heavy transformation, writing to a sink) in a span, tagging each span with attributes that make it useful for debugging (row counts, partition info, duration), and emitting metrics for the job's overall throughput and success alongside the traces, so you get both the aggregate trend and the ability to drill into one specific run's stage-by-stage behavior.

Structured elaboration

• Span structure for key operations: a parent span for the whole job run, with child spans for read_kafka (attributes: topic, partition count, records read), shuffle_stage (attributes: shuffle read/write bytes, partition count, skew indicator if available), and write_parquet (attributes: output path, rows written, file count), plus an explicit job_result attribute on the parent span capturing success or failure.
• Attributes to emit: at minimum, records_processed, duration_ms, and a stable job/run identifier on every span, plus stage-specific attributes (shuffle bytes for the shuffle stage, output row count for the write stage) that let you diagnose which stage is atypical without needing to re-run the job with verbose logging enabled.
• Sampling and cardinality trade-offs: full-fidelity tracing of every job run is usually affordable for a batch pipeline (runs are infrequent relative to a high-throughput streaming service), so sampling matters less here than tag cardinality does, avoid attaching an unbounded-cardinality attribute (a raw record id) directly as a span tag, keep tags to bounded-cardinality dimensions (job name, stage name, a coarse status) and push per-record detail into logs if needed.

Worked example

Concretely, a run of the orders_etl job produces a trace: parent span orders_etl_run (duration 18 min, job_result=success), child span read_kafka (duration 40s, records_read=2.1M), child span shuffle_stage (duration 15 min, unusually long relative to the job's typical 4-minute shuffle stage, shuffle_write_bytes=180GB, well above the typical 40GB), child span write_parquet (duration 90s, rows_written=2.05M). The trace immediately localizes today's slowdown to the shuffle stage specifically, and the elevated shuffle-write-bytes attribute (4.5x normal) points toward a join-key skew or a recent change increasing the join's fan-out, giving the on-call engineer a specific, actionable starting point rather than "the job ran slow today" with no further detail.

Trade-offs and pitfalls

Attaching stage-specific diagnostic attributes (shuffle bytes, not just duration) is what turns a trace from "which stage was slow" into "why was it slow," the duration alone tells you WHERE to look, but the attributes tell you WHAT to look for once you're there. The pitfall in cardinality is subtler for batch jobs than for high-throughput streaming: even though full-fidelity tracing of every RUN is cheap, attaching a per-record identifier as a span attribute (rather than an aggregate count) can still blow up cardinality if that identifier ends up indexed, so the discipline of keeping tags to bounded, aggregate-level attributes still applies even at low trace volume.