OpenTelemetry for Data Pipelines: Traces, Metrics & Logs Across Airflow, Spark & dbt

OpenTelemetry is the standard that finally lets a data platform stop stitching together three incompatible monitoring stacks — one for traces, one for metrics, one for logs — and instrument every pipeline once, with a single SDK and a single wire format, no matter whether the work runs in Airflow, Spark, or dbt. The hard problem in data-pipeline observability was never "collect some numbers"; it was that a single nightly run crosses a scheduler, a distributed compute engine, and a transformation tool, and each of those emitted its own telemetry into its own silo, so when the fact table landed late nobody could answer the only question that mattered — which task, in which run, was slow, and why. A metric told you latency rose; a log told you an exception fired; nothing told you they were the same run.

This guide is the senior-data-engineering walkthrough for wiring that single standard end to end, framed the way interviewers actually probe it: what a span is and how a whole pipeline run becomes one distributed trace, how W3C traceparent context propagates across an Airflow task → Spark stage → dbt model boundary so the trace does not fragment, which OTLP metric instruments to reach for and why delta-versus-cumulative temporality matters, how structured logs get correlated to the span that emitted them, and how the OTLP collector receives, samples, and fans everything out to whatever backend you choose without re-instrumenting a line of code. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the ETL practice library →, rehearse pipeline internals on the data processing practice library →, and sharpen the streaming axis with the streaming practice library →.

---

On this page

• Why OpenTelemetry changed data-pipeline observability
• Traces and spans across a data pipeline
• Metrics — pipeline SLIs with OTLP instruments
• Logs — structured and correlated to traces
• The OTLP Collector and instrumenting Airflow, Spark, and dbt
• Cheat sheet — OpenTelemetry for data pipelines
• Frequently asked questions
• Practice on PipeCode

---

1. Why OpenTelemetry changed data-pipeline observability

Three signals, one SDK, one wire format — the choice that decides whether a late table is debuggable

The one-sentence invariant: OpenTelemetry is a single vendor-neutral standard that unifies the three telemetry signals — traces, metrics, and logs — behind one instrumentation API and one wire protocol (OTLP), so that one pipeline run emits correlated telemetry that a Collector can route to any backend, and the engineering decision is no longer "which monitoring vendor" but "which signal answers which question, and how does context survive the hop between Airflow, Spark, and dbt." The moment you instrument with the OTel SDK instead of a vendor agent, the backend becomes a config line in the Collector rather than a rewrite — and that decoupling is the whole point, because data platforms outlive monitoring vendors.

The four axes interviewers actually probe.

• Signal choice. Do you know which of the three signals answers which question? Traces answer "where did the time go in this run"; metrics answer "is the fleet healthy over time"; logs answer "what exactly happened in this one event." Reaching for a log to compute a p95, or a metric to debug a single run, is the tell of someone who has not internalised the model. Interviewers open here because signal choice drives cost, cardinality, and query shape.
• Context propagation. A data pipeline run is a distributed operation — scheduler, cluster, transformer — and the whole value of tracing collapses if the trace_id does not survive the boundary between them. The senior answer names W3C traceparent, explains injecting it into XCom / Kafka headers / environment, and knows that a dropped context turns one trace into three orphans.
• Cardinality and cost. Every attribute you attach to a metric or span is a cost multiplier. Putting user_id or a raw file path on a metric label explodes the time series count and bankrupts the backend. The senior answer talks about bounded attribute sets, resource attributes vs data-point attributes, and sampling.
• Backend portability. The reason to adopt OTel over a proprietary agent is that instrumentation and destination are decoupled: you emit OTLP, and the Collector exports to Prometheus, Tempo, Loki, Datadog, or three of them at once. Interviewers want to hear that you instrument once and treat the backend as swappable.

The 2026 reality — OTel is the observability standard; the tools are at different maturity.

• Traces, metrics, and logs are all stable in the OTel spec, and OTLP is the default export protocol across the ecosystem. The Collector is production-hardened and is the single most important operational component.
• Airflow ships native OpenTelemetry — OTel metrics since 2.7 and OTel traces for DAG runs and task instances since 2.10 — so a scheduler-level trace and task metrics are a config block, not custom code.
• Spark has no native OTel, so you attach the OpenTelemetry Java agent to the driver and executors for JVM auto-instrumentation, and add a SparkListener to turn job/stage/task events into spans and metrics.
• dbt has no native OTel emitter, so you wrap the invocation in a root span and synthesise one child span per model from run_results.json timings — or lean on the emerging community adapters and OpenLineage bridges.

What interviewers listen for.

• Do you name traces, metrics, and logs as three distinct signals with distinct jobs — not "logging and monitoring"? — senior signal.
• Do you say traceparent and "propagate context across the task boundary" unprompted when tracing comes up? — required answer.
• Do you flag cardinality as the thing that decides cost before anyone asks about the bill? — senior signal.
• Do you describe the Collector as the decoupling layer that makes the backend swappable? — required answer.
• Do you resist "just use Prometheus and grep the logs" by explaining what a trace gives you that neither does? — senior signal.

Worked example — the three-signal decision table

Detailed explanation. The single most useful artifact for an observability interview is a memorised mapping of question → signal → instrument. Every senior telemetry discussion converges on it: given a symptom, which signal do you open first? Walk through building the table for a nightly Airflow DAG that triggers a Spark job and then runs dbt models.

• The run. daily_sales DAG: extract (Airflow task) → transform (Spark job) → dbt build (three models) → publish.
• The symptoms. "The dashboard was 40 minutes late." "Failures spiked this week." "This one run produced wrong numbers."
• The signals. Traces for the single-run latency breakdown; metrics for the week-over-week trend; logs for the single-event forensics.

Question. For each symptom, name the signal you open first and the specific instrument or query you run.

Input.

Symptom	Signal	Concretely
"Last night's run was 40 min late"	traces	open the run's trace; find the longest span
"Failure rate is up this week"	metrics	plot the pipeline.task.failures counter by day
"Row counts look wrong in one run"	logs	filter logs by that run's trace_id; read the transform log
"p95 task duration crept up"	metrics	query the task.duration histogram p95
"Which task caused the 40 min?"	traces	the span waterfall shows the critical path

Code.

# One instrumented Airflow task emits all three signals for the same run.
from opentelemetry import trace, metrics
import logging

tracer = trace.get_tracer("daily_sales.extract")
meter  = metrics.get_meter("daily_sales.extract")
log    = logging.getLogger("daily_sales.extract")

rows_counter = meter.create_counter(
    "pipeline.rows.extracted",
    unit="{row}",
    description="rows read from the source this task run",
)

def extract(**context):
    with tracer.start_as_current_span("extract_orders") as span:
        span.set_attribute("pipeline.name", "daily_sales")
        span.set_attribute("source.table", "public.orders")
        rows = read_source_rows()                 # your extract logic
        rows_counter.add(len(rows), {"source.table": "public.orders"})
        log.info("extracted rows", extra={"row_count": len(rows)})
        span.set_attribute("pipeline.rows", len(rows))
        return rows

Step-by-step explanation.

1. One task, one start_as_current_span — this is the trace signal. The span carries pipeline.name and source.table as attributes so the run is filterable, and its duration is measured automatically by the SDK when the with block exits.
2. rows_counter.add(...) is the metric signal. It is a monotonic Counter, so the backend can rate it (rows/min) and trend it across every run — the right tool for "is throughput drifting over the week?"
3. log.info(...) is the log signal. Because the log call happens inside the active span, the OTel logging bridge (section 4) stamps the record with the span's trace_id and span_id, so this single-event message is joinable back to the exact run.
4. The same len(rows) value appears in all three signals for a reason: the span attribute lets you see the count on the waterfall, the counter lets you trend it, and the log lets you audit one specific run. Each signal serves a different question about the same fact.
5. Reaching for the wrong signal is the classic mistake: computing a weekly failure rate by grepping logs (expensive, unbounded), or debugging a single slow run from a metric dashboard (aggregates hide the individual). The table is the antidote.

Output.

Question shape	Right signal	Wrong signal (common mistake)
"Where did the time go in this run?"	trace waterfall	staring at a metric graph
"Is the trend getting worse?"	metric time series	eyeballing individual traces
"What exactly happened in this event?"	correlated log	a metric counter
"Are these the same run?"	shared trace_id	guessing by timestamp

Rule of thumb. Match the signal to the question: traces for one run's latency breakdown, metrics for fleet trends over time, logs for single-event forensics. When someone debugs a trend with traces or a single run with metrics, they picked the wrong tool.

Worked example — what interviewers actually probe

Detailed explanation. The senior observability interview has a predictable escalation: an ambiguous opener ("how would you make this pipeline observable?"), then progressive narrowing to test whether you know the three signals, context propagation, and cost. The candidates who name the signal-to-question mapping and say "propagate traceparent" score highest.

• Ambiguous opener. "This Airflow → Spark → dbt pipeline is a black box. How do you make it observable?"
• Follow-up 1. "A run was slow. Walk me through debugging it." — probes traces + critical path.
• Follow-up 2. "How does the trace stay connected across Spark?" — probes context propagation.
• Follow-up 3. "Won't this blow up your metrics bill?" — probes cardinality.
• Follow-up 4. "Why OTel and not just Prometheus + ELK?" — probes portability + correlation.

Question. Draft a 5-minute senior observability answer that pre-empts all four follow-ups without waiting to be asked.

Input.

Interview signal	Weak answer	Senior answer
Instrumentation	"add more logging"	"instrument with the OTel SDK: traces, metrics, logs"
Debugging a slow run	"check the Airflow logs"	"open the run's trace; the span waterfall shows the critical path"
Cross-tool context	"each tool has its own dashboard"	"propagate W3C traceparent so it's one trace"
Cost	"we'll deal with the bill later"	"bound attribute cardinality; sample tails"
Vendor	"we use Datadog"	"emit OTLP; the Collector exports anywhere"

Code.

Senior OpenTelemetry answer template (5 minutes)
================================================

Minute 1 — name the model up front
  "I'd instrument with OpenTelemetry: three signals — traces for
   per-run latency, metrics for fleet SLIs, logs for event forensics —
   all through one SDK, exported as OTLP."

Minute 2 — traces + critical path
  "A pipeline run is one trace. Each Airflow task, Spark stage, and dbt
   model is a span. To debug a slow run I open its trace and read the
   waterfall: the longest span on the critical path is the culprit."

Minute 3 — context propagation
  "The trace only stays whole if context survives the hop. I inject the
   W3C traceparent header — into XCom between Airflow tasks, into the
   Spark submit env, into dbt's invocation — so the child spans attach
   to the right parent instead of starting three orphan traces."

Minute 4 — metrics + cardinality
  "SLIs are metrics: a rows-processed counter, a task-duration
   histogram, a freshness-lag gauge. I keep attributes bounded — table
   name yes, user_id never — so the time-series count stays sane and
   the bill doesn't explode."

Minute 5 — Collector + portability
  "Everything exports OTLP to a Collector. It batches, applies tail
   sampling, and fans out to whatever backend — Prometheus, Tempo,
   Loki, a vendor. Swapping backends is a Collector config change, not
   a re-instrumentation. That decoupling is why OTel beats a
   vendor-specific agent."

Step-by-step explanation.

1. Minute 1 frames the whole answer around three signals, one SDK. Weak candidates say "add logging"; naming traces/metrics/logs as distinct tools signals you understand the model, not just the tooling.
2. Minute 2 shows you debug a single run with a trace, and specifically with the critical path — the longest chain of spans — rather than scrolling logs. This is the single most senior thing you can say about tracing.
3. Minute 3 pre-empts the propagation follow-up. Naming traceparent and the exact injection points (XCom, Spark env, dbt invocation) is the difference between "I've read about tracing" and "I've wired it across tools."
4. Minute 4 pre-empts the cost follow-up. Volunteering the cardinality rule ("table yes, user_id never") before the interviewer raises the bill shows you have operated this in production.
5. Minute 5 closes on portability — the strategic reason OTel exists. "Instrument once, swap the backend in Collector config" is the sentence that separates a standard-adopter from a vendor-locked engineer.

Output.

Grading criterion	Weak score	Senior score
Names three signals distinctly	rare	mandatory
Debugs a run via trace waterfall	occasional	mandatory
Names traceparent propagation	rare	senior signal
Volunteers cardinality control	rare	senior signal
Frames Collector as portability	rare	senior signal

Rule of thumb. The senior OTel answer is a 5-minute monologue that covers signal choice, propagation, cardinality, and portability without waiting for the follow-ups. Rehearse it once; deploy it every interview.

Worked example — the "OTel vs Prometheus + ELK" pushback

Detailed explanation. A common interview trap is "we already have Prometheus for metrics and ELK for logs — why add OpenTelemetry?" The weak answer is "OTel is newer." The senior answer explains what the split stacks cannot do: correlate a metric spike to the exact trace to the exact log, and stay vendor-neutral. Walk the comparison.

• The status quo. Prometheus scrapes metrics; Logstash ships logs to Elasticsearch; there are no traces and no shared identifier across the two.
• The gap. A latency spike in Prometheus and an error in Kibana cannot be proven to be the same run — you correlate by eyeballing timestamps.
• The OTel answer. One SDK emits all three signals sharing a trace_id; an exemplar links a metric bucket to a trace; a log carries the trace_id. Correlation becomes a click, not a guess. And OTLP keeps Prometheus/Elastic as possible exporters, not a lock-in.

Question. Contrast the split Prometheus + ELK stack with an OTel-instrumented pipeline on correlation, tracing, and portability.

Input.

Capability	Prometheus + ELK	OpenTelemetry
Metrics	yes (Prometheus)	yes (OTLP → Prometheus/any)
Logs	yes (ELK)	yes (OTLP → Loki/Elastic/any)
Traces	no first-class tracing	yes (spans, native)
Metric↔log↔trace correlation	manual, by timestamp	shared trace_id + exemplars
Vendor lock-in	per-tool agents/config	instrument once, export anywhere

Code.

# The correlation OTel gives you: an exemplar ties a slow histogram
# bucket to the exact trace, and the log carries the same trace_id.
from opentelemetry import trace, metrics
import logging

tracer = trace.get_tracer("transform")
meter  = metrics.get_meter("transform")
log    = logging.getLogger("transform")

dur = meter.create_histogram("pipeline.task.duration", unit="s")

def transform_partition(part):
    with tracer.start_as_current_span("transform_partition") as span:
        ctx = span.get_span_context()
        span.set_attribute("partition.id", part.id)
        elapsed = run_transform(part)
        # Recording inside the active span lets the SDK attach an
        # exemplar carrying this trace_id to the histogram bucket.
        dur.record(elapsed, {"stage": "transform"})
        if elapsed > 30:
            log.warning(
                "slow partition",
                extra={"trace_id": format(ctx.trace_id, "032x")},
            )

Step-by-step explanation.

1. Prometheus + ELK can each answer their own question well, but they share no identifier. A p95 spike and an error log are correlated by hand, matching wall-clock timestamps across two UIs — slow and error-prone under load.
2. With OTel, dur.record(...) executes inside the active span, so the SDK can attach an exemplar — a sample data point carrying the current trace_id — to the histogram bucket. Clicking the slow bucket jumps straight to the offending trace.
3. The log.warning(...) carries the same trace_id, so the log, the metric bucket, and the trace all point at one run. That three-way join is the capability the split stack structurally cannot provide.
4. Crucially, adopting OTel does not throw away Prometheus or Elastic — they become exporters behind the Collector. You keep the backends you like and gain correlation and tracing on top.
5. The portability payoff: the instrumentation code above never names a vendor. Moving from self-hosted Prometheus to a managed backend is a Collector config edit, not a code change across every pipeline.

Output.

Question	Prometheus + ELK	OpenTelemetry
"Is this metric spike the same run as this error?"	guess by timestamp	click the exemplar
"Where did the run's time go?"	no trace to open	span waterfall
"Switch metrics backend"	re-configure scrape + agents	one Collector export line
"Add a new signal later"	new stack	already unified

Rule of thumb. OpenTelemetry does not replace Prometheus and ELK so much as unify and decouple them — it adds first-class traces, gives all three signals a shared trace_id for one-click correlation, and turns the backend into swappable Collector config. Frame it as "instrument once, correlate everything, keep the backends optional."

Senior interview question on data-pipeline observability strategy

A senior interviewer often opens with: "You inherit an Airflow → Spark → dbt pipeline with no tracing, ad-hoc print logging, and a Prometheus dashboard nobody trusts. Design an OpenTelemetry rollout: which signals you add first, how you keep it one trace across the three tools, how you stop the metrics bill from exploding, and how you avoid re-instrumenting when the backend changes."

Solution Using a phased OTel rollout with a Collector, bounded cardinality, and traceparent propagation

# Step 1 — a shared bootstrap module every pipeline imports.
# Configures the SDK once; everything exports OTLP to the Collector.
from opentelemetry import trace, metrics
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter

def init_otel(service_name: str, pipeline: str) -> None:
    # Resource attributes describe WHO is emitting; low cardinality, set once.
    resource = Resource.create({
        "service.name": service_name,     # e.g. "airflow-worker"
        "pipeline.name": pipeline,        # e.g. "daily_sales"
        "deployment.environment": "prod",
    })

    # Traces → Collector via OTLP/gRPC, batched.
    tp = TracerProvider(resource=resource)
    tp.add_span_processor(BatchSpanProcessor(
        OTLPSpanExporter(endpoint="otel-collector:4317")))
    trace.set_tracer_provider(tp)

    # Metrics → Collector, exported every 15s.
    reader = PeriodicExportingMetricReader(
        OTLPMetricExporter(endpoint="otel-collector:4317"),
        export_interval_millis=15000)
    metrics.set_meter_provider(MeterProvider(resource=resource, metric_readers=[reader]))

# Step 2 — propagate context across the Airflow task boundary via XCom.
from opentelemetry import trace
from opentelemetry.propagate import inject, extract

tracer = trace.get_tracer("daily_sales")

def upstream_task(ti, **_):
    with tracer.start_as_current_span("extract") as span:
        carrier: dict[str, str] = {}
        inject(carrier)                       # write traceparent into carrier
        ti.xcom_push(key="otel_ctx", value=carrier)

def downstream_task(ti, **_):
    carrier = ti.xcom_pull(key="otel_ctx", task_ids="upstream_task")
    parent_ctx = extract(carrier)             # rebuild parent context
    with tracer.start_as_current_span("load", context=parent_ctx):
        run_load()

# Step 3 — Collector caps cost and fans out; swap backends here, not in code.
processors:
  memory_limiter: { check_interval: 1s, limit_mib: 1500 }
  batch: { send_batch_size: 8192, timeout: 5s }
  tail_sampling:
    policies:
      - name: keep-errors
        type: status_code
        status_code: { status_codes: [ERROR] }
      - name: keep-slow
        type: latency
        latency: { threshold_ms: 30000 }
      - name: sample-rest
        type: probabilistic
        probabilistic: { sampling_percentage: 5 }
service:
  pipelines:
    traces:  { receivers: [otlp], processors: [memory_limiter, tail_sampling, batch], exporters: [otlp/tempo] }
    metrics: { receivers: [otlp], processors: [memory_limiter, batch], exporters: [prometheus] }
    logs:    { receivers: [otlp], processors: [memory_limiter, batch], exporters: [otlp/loki] }

Step-by-step trace.

Step	Before (no OTel)	After (phased OTel)
Debug a slow run	scroll three UIs, match timestamps	open one trace; read the waterfall
Cross-tool identity	none	shared trace_id via traceparent
Metrics cost	unbounded labels, surprise bill	bounded resource attrs; 5% tail sample
Add tracing	not possible	spans already flowing
Change backend	re-agent every host	one Collector export line
Rollout order	—	signals in: traces → metrics → logs

After the rollout, init_otel runs once per process so every Airflow worker, Spark driver, and dbt wrapper emits OTLP to the same Collector. The traceparent injected into XCom keeps the DAG a single trace; tail sampling keeps 100% of error and slow traces while sampling 5% of the boring ones; and the backend choices live entirely in the Collector's exporters, so a vendor migration never touches pipeline code.

Output:

Metric	Before	After
Mean time to locate a slow task	30–60 min across UIs	< 2 min on the waterfall
Trace continuity across tools	0% (three silos)	one trace per run
Traces retained	all or nothing	100% error/slow + 5% baseline
Backend migration effort	re-instrument all hosts	one config change
New-signal onboarding	new stack per signal	already unified

Why this works — concept by concept:

• OpenTelemetry SDK + OTLP — instrumentation and destination are decoupled. Code emits the vendor-neutral OTLP wire format; the backend is chosen downstream, so pipelines never encode a vendor.
• Resource attributes — service.name, pipeline.name, and deployment.environment are low-cardinality identifiers set once per process. They describe who is emitting without multiplying time series the way per-row attributes would.
• traceparent propagation via inject/extract — inject serialises the current context into a carrier that rides through XCom; extract rebuilds it on the other side so the downstream span attaches to the right parent. Without it, each task starts a fresh orphan trace.
• tail_sampling in the Collector — decisions are made after the whole trace is seen, so every error and every slow trace is kept while boring baseline traces are sampled at 5%. Cost drops without losing the traces you actually debug.
• Cost — one Collector deployment (agent + gateway), ~1–3% overhead per instrumented process, and a bounded time-series count from disciplined attributes. The eliminated cost is the multi-UI, timestamp-matching debugging tax on every incident — O(1) correlation via trace_id versus O(humans) guesswork.

Design
Topic — design
Design problems on observability and monitoring systems

Practice →

Data processing Topic — data-processing Data processing problems on pipeline instrumentation

Practice →

---

2. Traces and spans across a data pipeline

A span is one unit of work; a pipeline run is one trace — and the trace is only whole if context survives every boundary

The mental model in one line: a trace is a tree of spans, where each span records one unit of work with a start time, a duration, a trace_id shared across the whole run, its own span_id, a pointer to its parent, plus attributes, events, and a status — and in a data pipeline the natural mapping is one root span per run with a child span for every Airflow task, Spark stage, and dbt model, which only stays connected if the W3C traceparent context is propagated across each tool boundary instead of each tool starting its own orphan trace. Get the tree right and a slow run is a glance at a waterfall; get propagation wrong and you have three disconnected traces that answer nothing.

The anatomy of a span.

• Identity. Every span carries a 16-byte trace_id (shared by every span in the run) and an 8-byte span_id (unique to it). A child span also stores its parent's span_id, which is what builds the tree.
• Timing. A span has a start timestamp and an end timestamp; the SDK computes the duration when the span ends. A span you forget to end never closes and pollutes the trace.
• Attributes. Key–value pairs describing this unit of work — pipeline.name, source.table, spark.stage.id, dbt.model. Attributes are how you filter and group traces.
• Events and status. A span can carry timestamped events (e.g. "retry attempted") and a status (OK, ERROR, UNSET). Recording an exception sets the status to ERROR and attaches the stack as an event.

Mapping a data pipeline onto a span tree.

• Root span = the run. Start a root span when the DAG run begins (pipeline_run); everything else nests under it. Its trace_id is the run's identity for the whole telemetry lifetime.
• One span per task. Each Airflow task is a child span. Inside a task that does several things, add nested spans (read_source, write_staging) so the waterfall shows where within the task time went.
• Spark: job → stage → task spans. A SparkListener turns Spark's own job/stage/task events into spans, so a single transform span expands into the stage tree that actually consumed the wall clock.
• dbt: one span per model. Synthesise a child span per model from run_results.json timings, so a slow dbt build decomposes into the exact model that dragged.

W3C traceparent — the wire format that keeps the tree whole.

• What it is. A single header string, traceparent: 00-<trace_id>-<span_id>-<flags>, that encodes the current context. It is the standard every OTel SDK reads and writes.
• Propagation API. inject(carrier) writes traceparent into a dict/headers; extract(carrier) rebuilds a context you pass as the context= of the next span. That pair is the whole mechanism.
• Carriers in data pipelines. Airflow → Airflow: XCom. Airflow → Spark: an env var or --conf on submit. Producer → consumer: Kafka record headers. dbt: an environment variable read by an on-run-start hook.
• The failure if you skip it. Each tool calls start_as_current_span with no parent context, so it mints a new trace_id — one logical run becomes three traces and the waterfall you needed never exists.

The three failure modes senior engineers pre-empt.

• Broken parent links. Context not propagated (or propagated into the wrong carrier key) so child spans orphan. Mitigation: a shared helper that always injects on push and extracts on pull; assert the traceparent key exists.
• Unended spans. A span started without a with block (or a manually-started span whose end() is skipped on an exception path) never closes. Mitigation: prefer context managers; if starting manually, end in a finally.
• Sampling drops the parent. With head sampling, a child can be kept while its parent was dropped, producing a rootless fragment. Mitigation: use ParentBased samplers so a sampling decision propagates with the context, or move the decision to Collector tail sampling.

Common interview probes on traces.

• "What's the difference between a trace and a span?" — a trace is the whole tree for one run; a span is one node/unit of work.
• "How do you keep one trace across Airflow, Spark, and dbt?" — propagate W3C traceparent across each boundary.
• "What's on the critical path?" — the longest chain of spans whose durations sum to the trace duration; that's what to optimise.
• "How do you record a failure on a span?" — record_exception + set status ERROR.

Worked example — a root span with nested child spans in an Airflow task

Detailed explanation. The canonical trace start: open a root span for the run, then nest child spans for each sub-step of a task so the waterfall shows where inside the task the time went. Build an extract task that reads a source, validates, and writes staging — three nested spans under one task span.

• Root. pipeline_run opened at DAG start (shown here inline for clarity).
• Task span. extract_orders.
• Child spans. read_source, validate, write_staging — each timed independently.

Question. Instrument an extract task so the trace shows the task duration and the breakdown across read, validate, and write.

Input.

Span	Parent	What it measures
pipeline_run	(root)	whole run
extract_orders	pipeline_run	the task
read_source	extract_orders	source read
validate	extract_orders	validation pass
write_staging	extract_orders	staging write

Code.

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer("daily_sales.extract")

def extract_orders():
    # In a real DAG the root is opened once at run start and propagated;
    # shown inline here so the nesting is visible in one place.
    with tracer.start_as_current_span("extract_orders") as task_span:
        task_span.set_attribute("pipeline.name", "daily_sales")
        task_span.set_attribute("source.table", "public.orders")

        with tracer.start_as_current_span("read_source") as s:
            rows = read_rows("public.orders")
            s.set_attribute("pipeline.rows", len(rows))

        with tracer.start_as_current_span("validate") as s:
            bad = [r for r in rows if r["total_cents"] < 0]
            s.set_attribute("validation.bad_rows", len(bad))
            if bad:
                s.add_event("negative totals found", {"count": len(bad)})

        with tracer.start_as_current_span("write_staging"):
            write_staging(rows)

        task_span.set_status(Status(StatusCode.OK))
        return len(rows)

Step-by-step explanation.

1. start_as_current_span("extract_orders") opens the task span and makes it the current span, so any span opened inside automatically becomes its child — that is how nesting is established without passing IDs by hand.
2. Each inner with opens and closes a child span; the SDK stamps start/end automatically, so read_source, validate, and write_staging each get their own measured duration on the waterfall.
3. Attributes are attached at the level they describe: pipeline.rows on read_source (how many rows that step saw), validation.bad_rows on validate. This keeps each span's attributes about its own work.
4. add_event(...) records a timestamped marker inside the span — cheaper and more precise than a log for "this notable thing happened at this offset within the span."
5. Setting StatusCode.OK at the end makes the success explicit; on an exception path the with block still ends the spans, and you would record_exception + set ERROR (next example) so the failure shows on the trace.

Output.

Span	Duration	Key attribute
extract_orders	4.20 s	source.table=public.orders
read_source	3.10 s	pipeline.rows=120000
validate	0.30 s	validation.bad_rows=0
write_staging	0.80 s	—

Rule of thumb. Open a span per unit of work you might want to blame: one per task, and nested spans for the expensive sub-steps. If a step could ever be "the slow one," give it its own span so the waterfall can point at it.

Worked example — propagating traceparent from Airflow into a Spark job

Detailed explanation. The highest-value propagation hop in a data platform is Airflow → Spark, because Spark is usually where the time goes. The Airflow task injects traceparent into the Spark submit environment; the Spark driver extracts it and opens its job span as a child. Now the Spark stage tree hangs under the Airflow task in one trace.

• Airflow side. Inject the current context into a --conf / env var passed to spark-submit.
• Spark side. Read the env var, extract it, and open the driver's root span with that parent context.
• Result. Spark spans nest under the Airflow task span; one trace_id throughout.

Question. Wire traceparent from an Airflow task through spark-submit into the Spark driver so the Spark spans join the pipeline trace.

Input.

Boundary	Carrier	Mechanism
Airflow task → submit	env var TRACEPARENT	inject() on the Airflow side
submit → Spark driver	JVM reads env	extract() on the driver side
driver → stages	in-process context	SparkListener child spans

Code.

# Airflow side — inject context into the environment passed to spark-submit
from opentelemetry import trace
from opentelemetry.propagate import inject

tracer = trace.get_tracer("daily_sales")

def submit_spark(**_):
    with tracer.start_as_current_span("transform_spark") as span:
        span.set_attribute("spark.app", "daily_transform")
        carrier: dict[str, str] = {}
        inject(carrier)                       # {"traceparent": "00-<trace>-<span>-01"}
        env = {"TRACEPARENT": carrier["traceparent"]}
        run_spark_submit(app="daily_transform.py", env=env)

# Spark driver side — extract the parent and open the job span under it
import os
from opentelemetry import trace
from opentelemetry.propagate import extract

tracer = trace.get_tracer("daily_transform")

def main():
    carrier = {"traceparent": os.environ["TRACEPARENT"]}
    parent_ctx = extract(carrier)             # rebuild the Airflow task context
    with tracer.start_as_current_span("spark_job", context=parent_ctx) as job:
        job.set_attribute("spark.app.name", "daily_transform")
        spark = build_session_with_otel_listener()   # registers a SparkListener
        df = spark.read.parquet("s3://staging/orders/")
        df.groupBy("region").sum("total_cents").write.parquet("s3://curated/")

Step-by-step explanation.

1. On the Airflow side, inject(carrier) serialises the current span's context into a traceparent string. Because it runs inside transform_spark, the string points at that span as the parent-to-be.
2. The traceparent is passed to Spark as an ordinary environment variable — a carrier that survives the process boundary. Kafka headers or a --conf key work identically; the carrier is just "somewhere the string rides across the gap."
3. On the driver, extract(carrier) rebuilds a context object from the string. Passing it as context=parent_ctx tells the SDK "this new span's parent lives in that context" — even though the parent span object is in a different process.
4. Now spark_job is a child of the Airflow transform_spark span in the same trace: same trace_id, correct parent link. The SparkListener registered on the session opens further child spans per stage/task under spark_job.
5. Skip step 3 and the driver would call start_as_current_span with no parent, minting a brand-new trace_id — the Spark work would be a separate orphan trace and the Airflow waterfall would show a mysterious gap where Spark ran.

Output.

Trace assembled	trace_id	Parent chain
pipeline_run	4bf9…	(root)
↳ transform_spark (Airflow)	4bf9…	pipeline_run
↳ spark_job (driver)	4bf9…	transform_spark
↳ stage 0, stage 1 (listener)	4bf9…	spark_job

Rule of thumb. Treat every process boundary as a propagation checkpoint: inject on the way out, extract on the way in, and pass the extracted context as the context= of the first span on the far side. The carrier (env var, Kafka header, XCom) is interchangeable; the inject/extract discipline is not.

Worked example — recording failures and reading the critical path

Detailed explanation. A trace earns its keep on the bad night. When a task throws, record the exception on its span and set the status to ERROR so the span goes red and carries the stack. Then, to know what to fix, read the critical path — the chain of spans whose durations add up to the trace's wall-clock time. Optimising a span off the critical path buys nothing.

• Failure recording. record_exception(e) attaches the stack as a span event; set_status(ERROR) marks the span failed.
• Critical path. Not "the longest single span" but the longest dependency chain; parallel spans off the path do not extend the run.

Question. Instrument a task to record failures on its span, then identify the critical path of a run whose spans partly overlap.

Input.

Span	Start	End	On critical path?
pipeline_run	0.0	9.0	root
extract	0.0	3.0	yes
transform_spark	3.0	8.0	yes
dbt_build	8.0	9.0	yes
emit_metrics (parallel)	3.0	3.4	no (off path)

Code.

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer("daily_sales.transform")

def transform():
    with tracer.start_as_current_span("transform_spark") as span:
        try:
            run_spark()
        except Exception as e:
            span.record_exception(e)                       # stack as span event
            span.set_status(Status(StatusCode.ERROR, str(e)))
            raise                                          # let Airflow mark the task failed
        else:
            span.set_status(Status(StatusCode.OK))

Step-by-step explanation.

1. The try/except/else wraps the real work. On success the else sets OK; on failure the except records the exception and sets ERROR, then re-raises so Airflow still fails the task — telemetry must never swallow the error it is reporting.
2. record_exception(e) attaches a span event containing the exception type, message, and stack trace at the exact timestamp it happened, so the trace shows when within the span the failure occurred.
3. set_status(ERROR) turns the span red in the UI and, critically, lets Collector tail sampling keep this trace (the keep-errors policy from section 1) even if baseline traces are sampled away.
4. Reading the critical path: extract (0–3) → transform_spark (3–8) → dbt_build (8–9) form a dependency chain summing to the full 9 s. emit_metrics (3.0–3.4) runs in parallel and ends long before its siblings, so shaving it saves nothing.
5. The senior move is optimising on the path: transform_spark owns 5 of 9 seconds on the critical path, so it is the only span whose speedup shortens the run. Traces make this obvious; a metric dashboard hides it.

Output.

Span	Duration	Critical-path share	Worth optimising?
extract	3.0 s	33%	maybe
transform_spark	5.0 s	56%	yes — biggest lever
dbt_build	1.0 s	11%	low priority
emit_metrics	0.4 s	0% (parallel)	no — off path

Rule of thumb. Record exceptions and set ERROR on the span so failures are visible and always sampled; then optimise only what sits on the critical path. A faster span that runs in parallel with a slower sibling does not make the run any faster.

Senior interview question on distributed tracing for pipelines

A senior interviewer might ask: "A daily_sales DAG spans Airflow, a Spark job, and a dbt build, but your tracing backend shows three separate traces per run instead of one. Diagnose why, then design the propagation so it's a single trace — cover the carrier at each boundary, how you attach the Spark stage spans, how dbt models become spans, and how you record and always-sample failures."

Solution Using end-to-end traceparent propagation with a Spark listener and dbt run_results spans

# 1. Airflow: open the root span at run start, store its context in XCom,
#    and have every task extract it so the whole DAG is one trace.
from opentelemetry import trace
from opentelemetry.propagate import inject, extract

tracer = trace.get_tracer("daily_sales")

def start_run(ti, **_):
    span = tracer.start_span("pipeline_run")               # long-lived root
    span.set_attribute("pipeline.name", "daily_sales")
    carrier: dict[str, str] = {}
    with trace.use_span(span, end_on_exit=False):
        inject(carrier)
    ti.xcom_push(key="root_ctx", value=carrier)            # share with all tasks

def task_with_parent(ti, task_ids, name, fn):
    carrier = ti.xcom_pull(key="root_ctx", task_ids="start_run")
    parent = extract(carrier)
    with tracer.start_as_current_span(name, context=parent):
        fn()

# 2. Spark: a SparkListener converts job/stage events into child spans
#    under the driver's job span (which was extracted from TRACEPARENT).
from pyspark import SparkContext
from opentelemetry import trace

tracer = trace.get_tracer("daily_transform")

class OTelSparkListener:
    """Registered via spark._jsc; opens a span per stage, ends on completion."""
    def __init__(self):
        self._stage_spans = {}
    def onStageSubmitted(self, stage_info):
        sid = stage_info.stageId()
        self._stage_spans[sid] = tracer.start_span(f"spark.stage.{sid}")
    def onStageCompleted(self, stage_info):
        sid = stage_info.stageId()
        span = self._stage_spans.pop(sid, None)
        if span:
            span.set_attribute("spark.stage.tasks", stage_info.numTasks())
            span.end()

# 3. dbt: after `dbt build`, turn run_results.json timings into one span per model.
import json
from datetime import datetime
from opentelemetry import trace
from opentelemetry.propagate import extract
import os

tracer = trace.get_tracer("dbt")

def emit_dbt_spans():
    parent = extract({"traceparent": os.environ["TRACEPARENT"]})
    results = json.load(open("target/run_results.json"))
    with tracer.start_as_current_span("dbt_build", context=parent):
        for r in results["results"]:
            t = r["timing"][0]
            start = datetime.fromisoformat(t["started_at"].replace("Z", "+00:00"))
            end   = datetime.fromisoformat(t["completed_at"].replace("Z", "+00:00"))
            span = tracer.start_span(f"dbt.model.{r['unique_id']}",
                                     start_time=int(start.timestamp() * 1e9))
            span.set_attribute("dbt.status", r["status"])
            span.set_attribute("dbt.rows_affected", r.get("adapter_response", {}).get("rows_affected", 0))
            span.end(end_time=int(end.timestamp() * 1e9))

Step-by-step trace.

Boundary	Carrier	Attach mechanism
run start	XCom root_ctx	root span injected once
task → task	XCom pull + extract	context=parent on each task span
Airflow → Spark	env TRACEPARENT	driver extract, then SparkListener
Spark stages	in-process	listener opens span per stage
Airflow → dbt	env TRACEPARENT	emit_dbt_spans after build
dbt models	run_results.json	one span per model with real timings

After the fix, start_run opens the root once and every task, the Spark driver, and the dbt post-hook all extract the same context — so the backend assembles one trace with the same trace_id from pipeline_run down to each dbt.model.* span. The three-orphan-traces symptom is gone because no downstream ever calls start_as_current_span without a parent context.

Output:

Assembled trace	Spans	trace_id
pipeline_run	1 root	4bf9…
Airflow tasks	4	4bf9…
Spark job + stages	1 + N	4bf9…
dbt build + models	1 + M	4bf9…
Total	one connected tree	one trace_id

Why this works — concept by concept:

• Single root span in XCom — one long-lived pipeline_run span, injected once and shared via XCom, gives every task the same parent context, so the DAG is one trace instead of one-trace-per-task.
• inject / extract at every boundary — the propagation contract is symmetric: serialise on the way out, rebuild on the way in, and pass the rebuilt context as context=. The carrier differs per hop (XCom, env var) but the discipline is identical.
• SparkListener child spans — Spark has no native OTel, so the listener bridges Spark's own job/stage lifecycle events into spans under the driver's job span, exposing the stage tree that actually consumes wall-clock time.
• dbt run_results spans — dbt does not emit OTel, but its run_results.json carries exact per-model started_at/completed_at, so you back-date spans with start_time/end_time to reconstruct a faithful model-level waterfall after the fact.
• Cost — a few hundred spans per run, batched by the SDK and tail-sampled at the Collector, so retained volume is dominated by error/slow traces. The eliminated cost is the "three disconnected traces" debugging dead-end — O(1) waterfall read versus O(tools) manual correlation.

ETL
Topic — etl
ETL problems on pipeline orchestration and lineage

Practice →

Data processing Topic — data-processing Data processing problems on Spark job stages

Practice →

---

3. Metrics — pipeline SLIs with OTLP instruments

Counters count, histograms distribute, gauges sample — pick the wrong instrument and the SLI lies

The mental model in one line: OpenTelemetry metrics expose a small set of instruments — a Counter for monotonic totals, an UpDownCounter for values that rise and fall, a Histogram for distributions you want percentiles from, and asynchronous Observable instruments (counter/gauge) that a callback samples at export time — and the senior skill is matching the instrument to the SLI (rows processed → Counter, task duration → Histogram, freshness lag → async Gauge) while ruthlessly bounding attribute cardinality, because every distinct attribute combination is a separate time series the backend must store forever. Choose the instrument for the question — "how many?", "how spread out?", "what's the current level?" — not by habit.

The instruments and what each answers.

• Counter (sync, monotonic). Only goes up; add(n). For totals you rate downstream: rows extracted, tasks run, bytes written, failures. "How many, cumulatively?"
• UpDownCounter (sync, non-monotonic). Rises and falls; add(+n / -n). For live levels you mutate: queue depth, in-flight tasks, open connections. "How many right now, by delta?"
• Histogram (sync). Records a value into buckets; record(v). For distributions where you need p50/p95/p99: task duration, batch size, row latency. "How is it spread — and what's the tail?"
• Observable Gauge / Observable Counter (async). A callback the SDK invokes at export time. For values you sample rather than sum: data freshness lag, watermark age, disk usage, current partition count. "What's the level at read time?"

Temporality — the setting that must match the backend.

• Cumulative. Each export reports the running total since process start. Prometheus wants this; it computes rates itself with rate().
• Delta. Each export reports only what changed since the last export. Many hosted backends and OTLP-native pipelines prefer delta because it survives process restarts cleanly.
• The trap. Sending cumulative to a delta-expecting backend (or vice versa) produces nonsense rates — counters that appear to reset, or deltas that look like totals. Set temporality explicitly on the OTLP metric exporter to match the destination.

Views and aggregation — shaping metrics before they leave.

• Views rename instruments, drop unwanted attributes, and override the aggregation. The single most useful view drops a high-cardinality attribute at the SDK so it never reaches the backend.
• Explicit histogram buckets. The default bucket boundaries rarely fit pipeline durations; a View sets boundaries tuned to your range (e.g. seconds to tens of minutes) so p95 is meaningful.
• Aggregation choice. Sum, LastValue, or ExplicitBucketHistogram — a View can force the right one per instrument.

Cardinality — the cost axis that ends careers.

• The rule. Time-series count ≈ product of the distinct values of every attribute. table (dozens) is fine; user_id (millions) or file_path (unbounded) multiplies your bill into the ground.
• Resource vs data-point attributes. Put stable identity (service.name, pipeline.name) on the resource, set once. Put only bounded, low-arity discriminators (stage, status, table) on the data point.
• Mitigation. Drop high-cardinality attributes in a View; bucket continuous values; never label with IDs — put IDs on spans (traces), where per-item detail belongs.

Exemplars — the bridge from a metric back to a trace.

• What they are. A histogram bucket can carry an exemplar: a sample recording captured inside an active span, tagged with that span's trace_id.
• Why they matter. Clicking a spike in the p99 duration jumps straight to a representative slow trace — metric-to-trace correlation with no timestamp guessing.
• How to get them. record() inside an active, sampled span; the SDK attaches the exemplar automatically when exemplar collection is enabled.

Common interview probes on metrics.

• "Counter vs Histogram — when each?" — Counter for totals you rate; Histogram when you need percentiles.
• "How do you measure data freshness?" — an async Observable Gauge sampling now() - max(event_time).
• "What blows up a metrics bill?" — unbounded attribute cardinality; put IDs on traces, not metric labels.
• "Delta or cumulative?" — match the backend: Prometheus cumulative, many OTLP-native/hosted backends delta.

Worked example — a rows-processed Counter and a task-duration Histogram

Detailed explanation. The two workhorse pipeline SLIs are throughput (a Counter you rate into rows/min) and latency distribution (a Histogram you read p95 from). Instrument one task with both, with bounded attributes and histogram buckets tuned to pipeline durations.

• Counter. pipeline.rows.processed, attribute table (low cardinality).
• Histogram. pipeline.task.duration in seconds, buckets tuned from 1 s to 30 min.
• View. Sets the histogram bucket boundaries and drops any stray high-cardinality attribute.

Question. Instrument a transform task to emit a rateable throughput Counter and a percentile-ready duration Histogram, with a View that fixes the buckets.

Input.

Instrument	Name	Attribute(s)	Read as
Counter	pipeline.rows.processed	table	rows/min via rate
Histogram	pipeline.task.duration	stage	p50/p95/p99
View	(on the histogram)	—	explicit buckets

Code.

import time
from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.view import View, ExplicitBucketHistogramAggregation

# View: pin histogram buckets to the pipeline's real duration range (seconds).
duration_view = View(
    instrument_name="pipeline.task.duration",
    aggregation=ExplicitBucketHistogramAggregation(
        [1, 5, 10, 30, 60, 120, 300, 600, 1200, 1800]),
)
# (duration_view is passed to MeterProvider(views=[duration_view], ...) at init)

meter = metrics.get_meter("daily_sales.transform")
rows  = meter.create_counter("pipeline.rows.processed", unit="{row}")
dur   = meter.create_histogram("pipeline.task.duration", unit="s")

def transform(table: str):
    t0 = time.monotonic()
    n = run_transform(table)
    rows.add(n, {"table": table})                 # bounded attr: table only
    dur.record(time.monotonic() - t0, {"stage": "transform"})
    return n

Step-by-step explanation.

1. The Counter rows.add(n, {"table": table}) is monotonic — it only accumulates. The backend derives throughput with a rate function; you never compute rows/min in the pipeline, you just count.
2. The only attribute on the Counter is table, whose distinct values number in the dozens. That keeps the time-series count bounded. Adding run_id or partition here would create a new series per run — a cardinality bomb.
3. The Histogram dur.record(...) drops the elapsed seconds into a bucket. Percentiles are computed from bucket counts, so the bucket boundaries decide whether p95 is meaningful.
4. The default histogram boundaries top out around 10 s — useless for a 20-minute Spark stage. The View overrides them to span 1 s → 30 min, so a slow task lands in a distinct high bucket instead of saturating the last default bucket.
5. stage is the histogram's only attribute — again low cardinality. The result is two SLIs that trend cleanly across every run without multiplying series.

Output.

SLI	Instrument	Example read
Throughput	pipeline.rows.processed	rate(...) = 42,000 rows/min
Latency p50	pipeline.task.duration	120 s
Latency p95	pipeline.task.duration	540 s
Latency p99	pipeline.task.duration	1,150 s

Rule of thumb. Count with a Counter and rate it downstream; distribute with a Histogram and tune the buckets to your real duration range. Keep attributes to a handful of low-cardinality discriminators — the moment an ID appears on a metric label, the bill is a time bomb.

Worked example — data freshness as an async Observable Gauge

Detailed explanation. Freshness — "how stale is the newest data downstream?" — is the SLI business stakeholders actually feel, and it is a sampled value, not something you sum. That makes it an asynchronous Observable Gauge: register a callback that, at each export, computes now() - max(event_time) and reports it. No per-event instrumentation, just a periodic sample.

• The metric. pipeline.freshness.lag_seconds — current staleness of the curated table.
• The callback. Queries max(event_time) and subtracts from now.
• Why async. The value has no natural "add" moment; it is a level you read on a schedule.

Question. Implement a freshness-lag SLI as an Observable Gauge whose callback samples the newest row's age.

Input.

Aspect	Choice
Instrument	Observable (async) Gauge
Metric	pipeline.freshness.lag_seconds
Sample source	SELECT max(event_time) FROM curated.orders
Attribute	table (bounded)

Code.

import time
from opentelemetry import metrics
from opentelemetry.metrics import Observation, CallbackOptions

meter = metrics.get_meter("daily_sales.freshness")

def freshness_callback(options: CallbackOptions):
    # Called by the SDK at each export interval; returns the current level.
    max_event = query_scalar(
        "SELECT extract(epoch FROM max(event_time)) FROM curated.orders")
    lag = time.time() - (max_event or 0)
    yield Observation(lag, {"table": "curated.orders"})

# Registering the callback creates the async gauge; no add()/record() calls.
meter.create_observable_gauge(
    "pipeline.freshness.lag_seconds",
    callbacks=[freshness_callback],
    unit="s",
    description="age of the newest row in the curated table",
)

Step-by-step explanation.

1. Freshness has no "event" to hook — it is a state, so a synchronous Counter/Histogram is the wrong tool. An Observable Gauge fits: the SDK pulls the value on its own schedule.
2. create_observable_gauge(..., callbacks=[freshness_callback]) registers the callback. There is no add/record anywhere in the pipeline; the value is produced only when the SDK exports (e.g. every 15 s).
3. The callback yields an Observation — the current lag plus the bounded table attribute. Yielding multiple observations (one per table) is how one callback covers several series while staying low-cardinality.
4. Because the callback runs at export time, a stalled pipeline shows rising lag automatically — no code path in the pipeline needs to fire; the absence of new rows makes max(event_time) stale and the gauge climbs.
5. This is the SLI you alert on: pipeline.freshness.lag_seconds{table="curated.orders"} > 3600 means the curated table is over an hour behind, which is exactly the business-visible symptom stakeholders complain about.

Output.

Time	max(event_time)	Reported lag
09:00 (healthy)	08:58	120 s
10:00 (healthy)	09:57	180 s
11:00 (stalled)	09:57	3,780 s → alert
11:15 (recovered)	11:14	60 s

Rule of thumb. Model levels you sample (freshness, lag, queue depth, disk) as asynchronous Observable Gauges with a callback, not synchronous instruments. The callback runs at export time, so a stalled pipeline makes the gauge climb on its own — no in-pipeline code needs to fire.

Worked example — killing a cardinality explosion with a View

Detailed explanation. The most common production metrics incident is a cardinality explosion: someone labels a Counter with run_id or file_path, and the time-series count goes from hundreds to millions, tipping the backend over. The fix is a View that drops the offending attribute at the SDK — before it ever hits the wire — plus the discipline of moving per-item detail to spans.

• The bug. pipeline.rows.processed labelled with run_id (new value every run) → unbounded series growth.
• The fix. A View that keeps only table and drops run_id.
• The lesson. Per-run/per-item identity belongs on traces, not metric labels.

Question. Given a Counter accidentally labelled with an unbounded run_id, write the View that bounds its cardinality, and explain where run_id should live instead.

Input.

Before	After
attrs: table, run_id	attrs: table
series count	dozens (per table)
run_id visibility	on the span, not the metric

Code.

from opentelemetry.sdk.metrics.view import View
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader

# View: for this instrument, keep ONLY the `table` attribute; drop everything
# else (including the accidental high-cardinality run_id).
bound_rows = View(
    instrument_name="pipeline.rows.processed",
    attribute_keys={"table"},          # allow-list — all other attrs are dropped
)

reader = PeriodicExportingMetricReader(otlp_metric_exporter, export_interval_millis=15000)
provider = MeterProvider(views=[bound_rows], metric_readers=[reader])

# run_id still matters for debugging — so put it on the trace, not the metric:
def transform(run_id: str, table: str):
    with tracer.start_as_current_span("transform") as span:
        span.set_attribute("run.id", run_id)      # per-run detail lives here
        n = run_transform(table)
        rows.add(n, {"table": table, "run_id": run_id})  # run_id dropped by the View

Step-by-step explanation.

1. The View with attribute_keys={"table"} is an allow-list: for pipeline.rows.processed, only table survives; run_id is stripped at the SDK, so the wire and backend never see it. Series count collapses back to "one per table."
2. Applying it in MeterProvider(views=[bound_rows], ...) means the fix is central — no need to hunt down every rows.add call site and edit its attributes.
3. The rows.add(..., {"table": ..., "run_id": ...}) call can still pass run_id; the View drops it. That is intentional: it lets the same call site feed both a bounded metric and (if you wanted) a differently-scoped view without code churn.
4. run_id genuinely matters for debugging — so it goes on the span as run.id. Traces are designed for per-item, high-cardinality detail; metrics are not. This is the core division of labour between the two signals.
5. The net effect: the throughput SLI trends cleanly at dozens of series, while per-run forensics stay available on the trace. Cost bounded, debuggability preserved.

Output.

Dimension	With run_id on metric	After the View
Series per table per day	~1 per run (unbounded)	1 (stable)
Backend storage trend	grows without bound	flat
run_id still findable?	yes, but ruinously	yes — on the span
Fix location	every call site	one View

Rule of thumb. When a metric's cardinality explodes, fix it centrally with a View allow-list rather than editing every call site, and move the per-run identity to a span attribute. Metrics are for bounded trends; traces are where unbounded, per-item detail belongs.

Senior interview question on pipeline SLIs and metric cost

A senior interviewer might ask: "Define the SLIs for a batch pipeline feeding a business-critical table, pick the OTLP instrument for each, and stop the metrics bill from exploding. Cover throughput, task-duration percentiles, freshness lag, and failure rate; explain delta vs cumulative temporality for your backend; and show how you'd cap cardinality when a teammate labels everything with run_id."

Solution Using the right instrument per SLI, tuned Views, and bounded cardinality

from opentelemetry import metrics
from opentelemetry.metrics import Observation, CallbackOptions
from opentelemetry.sdk.metrics.view import View, ExplicitBucketHistogramAggregation

# 1. Views: fix histogram buckets + allow-list attributes to bound cardinality.
VIEWS = [
    View(instrument_name="pipeline.task.duration",
         aggregation=ExplicitBucketHistogramAggregation(
             [1, 5, 10, 30, 60, 120, 300, 600, 1200, 1800])),
    View(instrument_name="pipeline.rows.processed", attribute_keys={"table"}),
    View(instrument_name="pipeline.task.failures",  attribute_keys={"task", "reason"}),
]

meter = metrics.get_meter("daily_sales")

# 2. One instrument per SLI, chosen by the question it answers.
rows     = meter.create_counter("pipeline.rows.processed", unit="{row}")   # throughput
duration = meter.create_histogram("pipeline.task.duration", unit="s")      # latency dist
failures = meter.create_counter("pipeline.task.failures", unit="{fail}")   # error rate

def freshness_cb(options: CallbackOptions):
    lag = current_time() - newest_event_time("curated.orders")
    yield Observation(lag, {"table": "curated.orders"})

meter.create_observable_gauge("pipeline.freshness.lag_seconds",             # freshness
                              callbacks=[freshness_cb], unit="s")

# 3. Emit inside a task; failures counted with a bounded `reason`, not a message.
def run_task(task: str, table: str):
    import time
    t0 = time.monotonic()
    try:
        n = do_work(table)
        rows.add(n, {"table": table})
    except TimeoutError:
        failures.add(1, {"task": task, "reason": "timeout"})   # bounded reason label
        raise
    except Exception:
        failures.add(1, {"task": task, "reason": "other"})
        raise
    finally:
        duration.record(time.monotonic() - t0, {"stage": task})

# 4. Match temporality to the backend on the OTLP metric exporter.
from opentelemetry.sdk.metrics.export import AggregationTemporality
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter

# Prometheus scrape path wants CUMULATIVE; many hosted OTLP backends want DELTA.
exporter = OTLPMetricExporter(
    endpoint="otel-collector:4317",
    preferred_temporality={
        # counters/histograms as DELTA for a delta-native backend:
        # (use CUMULATIVE instead if exporting straight to Prometheus)
    },
)

Step-by-step trace.

SLI	Instrument	Why this instrument
Throughput	Counter rows.processed	monotonic total; rate it downstream
Latency	Histogram task.duration	need p50/p95/p99 from buckets
Failure rate	Counter task.failures	count of a bounded event
Freshness	Observable Gauge freshness.lag_seconds	sampled level, no add moment
Cardinality	Views (allow-list)	drop run_id-style attrs at SDK
Temporality	exporter setting	match Prometheus (cumulative) / OTLP backend (delta)

After deployment, each SLI uses the instrument that matches its question, the failure Counter uses a bounded reason label (timeout/other) instead of the raw exception message, the Views keep the whole metric surface at a few dozen series, and the exporter's temporality matches the backend so rates render correctly. Freshness climbs on its own when the pipeline stalls because its callback samples at export time.

Output:

Metric	Value / behaviour
Throughput	rate(rows.processed) ≈ 42k rows/min
Duration p95	540 s (from tuned buckets)
Failure rate	rate(task.failures{reason="timeout"})
Freshness lag	120 s healthy → climbs when stalled
Series count	dozens (bounded by Views)
Rate correctness	intact (temporality matched)

Why this works — concept by concept:

• Instrument per question — Counter for totals you rate, Histogram for distributions you need percentiles from, Observable Gauge for sampled levels. Matching the instrument to the question is what makes the SLI truthful.
• Bounded reason label — the failure Counter uses reason ∈ {timeout, other}, not the raw exception text, so the error-rate series stays low-cardinality while the detail lives on the span that recorded the exception.
• Views as the cardinality firewall — allow-listing attributes and pinning histogram buckets at the SDK means cost and percentile-fidelity are controlled centrally, before anything reaches the backend.
• Temporality matched to the backend — cumulative for Prometheus's rate(), delta for OTLP-native/hosted backends. Mismatch produces counters that appear to reset or deltas that read as totals — set it explicitly.
• Cost — a handful of instruments and a few dozen bounded series per pipeline, exported every 15 s. The eliminated cost is the runaway time-series bill from ID-valued labels — the single most common way a metrics backend falls over.

Optimization
Topic — optimization
Optimization problems on latency and percentile SLIs

Practice →

Data validation Topic — data-validation Data validation problems on freshness and volume checks

Practice →

---

4. Logs — structured and correlated to traces

A log without a trace_id is a needle with no haystack coordinates

The mental model in one line: an OpenTelemetry log record is a structured event — timestamp, severity, a body, and typed attributes — that the logging bridge emits through the same SDK and OTLP wire as traces and metrics, and its decisive feature for a data pipeline is correlation: because the record is created while a span is active, the SDK stamps it with that span's trace_id and span_id, so a single log line is joinable to the exact run and the exact task that produced it, turning "grep 40 GB of text" into "click the failed span, read its three logs." Structured plus correlated is the whole game; an unstructured, context-free log is where debugging goes to die.

The anatomy of an OTel log record.

• Timestamp + observed timestamp. When the event happened, and when the collector observed it — useful when logs are buffered.
• Severity. A number (SeverityNumber) plus text (INFO, WARN, ERROR), normalised across languages so filters work uniformly.
• Body. The human message — ideally short and stable ("task failed"), with the variable parts in attributes, not interpolated into the body.
• Attributes + resource. Typed key–values on the record (row_count, retry) plus the resource (service.name, pipeline.name); and, when a span is active, the injected trace_id / span_id.

The logging bridge — how existing loggers become OTel logs.

• What it is. An OTel handler you attach to the language's standard logger (Python logging, Log4j, etc.), so existing log calls flow into the OTel pipeline without rewriting every call site.
• Trace-context injection. While a span is active, the bridge reads the current context and stamps trace_id/span_id onto each record automatically — no manual plumbing per log line.
• Dual-path caution. If you keep the old file/stdout handler and add the OTel handler, decide deliberately: usually the Collector reads stdout or the SDK exports directly — not both, or you double-count.

Correlation — the payoff that justifies structured logs.

• Span → logs. From a red span, list every log record sharing its span_id. That is the "what exactly happened here" forensic view.
• Log → trace. From an alarming log line, jump to its trace_id and open the whole run's waterfall to see the context around the failure.
• Metric → log via trace. An exemplar takes you from a metric spike to a trace; the trace's spans take you to their logs. All three signals meet at the trace_id.

Structured logging discipline for pipelines.

• Stable body, variable attributes. log.error("write failed", extra={"table": t, "rows": n}) — not log.error(f"write failed for {t} with {n} rows"). Stable bodies group; interpolated bodies are unsearchable noise.
• Severity hygiene. INFO for lifecycle, WARN for recoverable anomalies, ERROR for failures that page. Do not log at ERROR for things you retry successfully — it trains on-call to ignore alerts.
• No PII / secrets in the body. Log bodies fan out to a backend and often a vendor. Keep identifiers as hashed or bounded attributes; never dump raw rows, tokens, or connection strings.

The three failure modes senior engineers pre-empt.

• Logs without trace context. A log emitted outside any active span has no trace_id, so it cannot be correlated. Mitigation: ensure the span is active when you log; for background threads, propagate context explicitly.
• Double export. Both the SDK exporter and a Collector filelog receiver ingest the same lines → duplicate records and double the bill. Mitigation: pick one path per deployment.
• PII leakage. Interpolated bodies smuggle emails, tokens, and raw rows into the logging backend. Mitigation: structured attributes with an allow-list; redact at the handler.

Common interview probes on logs.

• "How do you correlate a log to a trace?" — the record carries trace_id/span_id injected by the bridge while the span is active.
• "Why structured over free-text?" — stable bodies + typed attributes are filterable; interpolated strings are not.
• "How do OTel logs reach the backend?" — SDK exporter → OTLP → Collector, or Collector tails stdout — pick one.
• "What must never go in a log body?" — PII, secrets, raw rows; keep bounded/hashed attributes only.

Worked example — bridging Python logging into OTel with trace context

Detailed explanation. The lowest-friction adoption path is to keep every existing logging call and attach an OTel LoggingHandler. Once attached, any log emitted inside an active span is automatically stamped with trace_id/span_id and shipped as an OTLP log record. Set it up and log inside a span.

• Setup. A LoggerProvider with an OTLP log exporter; attach a LoggingHandler to the root Python logger.
• Usage. Ordinary log.info/warning/error calls, inside a span, now carry trace context.
• Structure. Variable data goes in extra=, not the message string.

Question. Configure the OTel logging bridge and emit a structured, trace-correlated log from inside a task span.

Input.

Piece	Value
Provider	LoggerProvider + OTLP log exporter
Handler	LoggingHandler on root logger
Body	stable string
Variable data	extra={...} attributes

Code.

import logging
from opentelemetry import trace
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter

# 1. Wire the logging bridge: OTLP log exporter → root Python logger.
provider = LoggerProvider()
provider.add_log_record_processor(
    BatchLogRecordProcessor(OTLPLogExporter(endpoint="otel-collector:4317")))
handler = LoggingHandler(level=logging.INFO, logger_provider=provider)
logging.getLogger().addHandler(handler)

tracer = trace.get_tracer("daily_sales.load")
log = logging.getLogger("daily_sales.load")

def load(table: str, rows: list):
    with tracer.start_as_current_span("load") as span:      # span is active…
        span.set_attribute("target.table", table)
        try:
            n = write_rows(table, rows)
            # stable body + structured attributes; trace_id stamped automatically
            log.info("load complete", extra={"target.table": table, "rows": n})
        except Exception:
            log.error("load failed", extra={"target.table": table})
            raise

Step-by-step explanation.

1. The LoggingHandler attached to the root logger is the bridge: every stdlib logging call now also becomes an OTel log record shipped via OTLP — no call-site rewrite, existing log.info lines just start correlating.
2. Because log.info("load complete", ...) runs inside the active load span, the bridge reads the current context and stamps the record with the span's trace_id and span_id. That stamping is automatic — the whole reason to log inside the span.
3. The body is the stable string "load complete"; the variable data (target.table, rows) rides in extra= as typed attributes. Now you can filter "all load complete events" and facet by table — impossible with an interpolated f-string.
4. On the error path, log.error("load failed", ...) is emitted inside the same span, so the failure log shares the span's trace_id — the log and the (soon-to-be-ERROR) span are joined without any manual ID passing.
5. The result: from the load span you can list exactly these two possible records, and from either record you can open the full run trace. Correlation both directions, for free, from ordinary logging calls.

Output.

Log record	trace_id	span_id	Attributes
load complete (INFO)	4bf9…	a1b2…	target.table, rows=120000
load failed (ERROR)	4bf9…	a1b2…	target.table

Rule of thumb. Attach the OTel logging bridge once, then log inside the active span with a stable body and structured extra= attributes. The bridge stamps trace_id/span_id automatically — a log emitted outside a span loses that correlation and is far harder to debug.

Worked example — correlating a failed Spark task log to its span

Detailed explanation. The debugging superpower is round-tripping: start from a red span, pull its logs; or start from an error log, open its trace. Show both directions for a Spark task that failed on a bad partition, where the Spark span and the executor's log share a trace_id.

• Scenario. spark.stage.3 span goes ERROR; an executor logs a serialization failure on partition 47.
• Span → logs. Query logs WHERE span_id = <stage span>.
• Log → trace. From the error log's trace_id, open the run waterfall.

Question. Show the queries/steps that go from a failed Spark span to its logs and from the error log back to the full trace.

Input.

Direction	Start	Key	Result
Span → logs	red spark.stage.3 span	its span_id	executor error logs
Log → trace	executor ERROR log	its trace_id	whole run waterfall

Code.

# Executor-side: log the failure inside the stage span so it shares context.
from opentelemetry import trace
import logging

log = logging.getLogger("daily_transform.executor")

def process_partition(part, stage_span_ctx):
    # stage_span_ctx was propagated from the driver's SparkListener span
    with trace.use_span(stage_span_ctx):                 # re-activate the stage span
        try:
            transform(part)
        except Exception as e:
            # trace_id/span_id stamped from the active stage span
            log.error("partition failed", extra={"partition.id": part.id,
                                                  "error.kind": type(e).__name__})
            raise

# Backend queries — the two-way correlation the shared trace_id enables

# (1) Span → logs: from the red stage span, list its log records
LogQL/Loki:   {service_name="daily_transform"} | json | span_id="c3d4..."
Result: partition failed  partition.id=47  error.kind=SerializationError

# (2) Log → trace: from that error log's trace_id, open the run
Trace UI:     trace_id = 4bf9...
Result: pipeline_run → transform_spark → spark_job → spark.stage.3 (ERROR)
        …the whole run, with stage.3 red and partition 47 named on the log

Step-by-step explanation.

1. On the executor, trace.use_span(stage_span_ctx) re-activates the stage span's context (propagated from the driver's listener span). Now any log inside the block is stamped with that stage's trace_id/span_id.
2. The log.error("partition failed", ...) uses a stable body and structured attributes (partition.id=47, error.kind). Because a span is active, the record inherits the correlation IDs — no manual passing.
3. Direction one (span → logs): in the logs backend, filter span_id = "c3d4..." (the stage span's id). Exactly the records emitted under that span come back — here, the partition-47 serialization failure. That is "what happened inside this red span."
4. Direction two (log → trace): take the error log's trace_id and open it in the trace UI. The full run waterfall renders with spark.stage.3 red — you see the failure in the context of the whole run, including what ran before and after.
5. This round-trip is only possible because the log and the span share a trace_id. Without the bridge stamping it, the executor log would be an isolated line and you would be back to matching timestamps across two systems.

Output.

Step	Query	What you learn
Span → logs	span_id="c3d4…"	partition 47, SerializationError
Log → trace	trace_id="4bf9…"	whole run; stage.3 is the red node
Blame	—	one partition, one serializer bug
Time to root cause	—	minutes, not hours

Rule of thumb. Emit failure logs inside the active span (re-activating it on executors/threads if needed) so log and span share a trace_id. Then correlation is bidirectional: span_id filters logs to one unit of work, and trace_id opens the log's full run context.

Worked example — severity hygiene and keeping PII out of log bodies

Detailed explanation. Two disciplines separate a log stream you trust from one on-call mutes: correct severity (don't cry ERROR for a successful retry) and no PII in bodies (which fan out to backends and vendors). Fix a noisy, leaky logger.

• Severity bug. A retried-then-succeeded transient logs at ERROR, so the error rate is fiction and alerts get ignored.
• PII bug. The body interpolates a customer email and a raw row, leaking PII into the logging backend.
• The fix. WARN for recoverable retries, ERROR only for terminal failure; structured, allow-listed, hashed attributes instead of raw values in the body.

Question. Rewrite a noisy, PII-leaking logger to use correct severities and safe structured attributes.

Input.

Problem	Before	After
Transient retry	ERROR "retrying…"	WARN with retry attr
Terminal failure	ERROR (correct)	ERROR (kept)
Customer email in body	interpolated raw	hashed attribute
Raw row in body	dumped	row count, not contents

Code.

import hashlib, logging
from opentelemetry import trace

log = logging.getLogger("daily_sales.notify")
tracer = trace.get_tracer("daily_sales.notify")

def _hash(v: str) -> str:
    return hashlib.sha256(v.encode()).hexdigest()[:12]   # bounded, non-reversible

def notify(customer_email: str, rows: list):
    with tracer.start_as_current_span("notify") as span:
        for attempt in range(3):
            try:
                send(customer_email, rows)
                return
            except TransientError:
                # recoverable → WARN, not ERROR; no email in the body
                log.warning("notify retry", extra={"retry": attempt + 1,
                                                    "customer.hash": _hash(customer_email)})
        # exhausted retries → terminal → ERROR; still no PII, no raw rows
        log.error("notify failed", extra={"customer.hash": _hash(customer_email),
                                           "row_count": len(rows)})
        span.set_status(trace.Status(trace.StatusCode.ERROR))

Step-by-step explanation.

1. The transient path logs at WARN with a retry attribute, not ERROR. A retry that later succeeds is not an error; logging it as one inflates the error rate and trains on-call to ignore the signal.
2. Only the terminal path — retries exhausted — logs at ERROR and sets the span status to ERROR. Now the error rate reflects real failures, and tail sampling keeps exactly those traces.
3. PII never touches the body. The customer email is reduced to a bounded, non-reversible customer.hash attribute — enough to correlate "same customer across events" without storing the address in the logging backend.
4. The raw rows are never dumped; only row_count is logged. Dumping row contents would smuggle real customer data (and huge payloads) into logs, a compliance and cost problem at once.
5. Bodies stay stable ("notify retry", "notify failed"), so they group cleanly and are searchable; all variability is in typed, allow-listed attributes. The stream is now trustworthy and privacy-safe.

Output.

Event	Severity	Body	Attributes
retry #1	WARN	notify retry	retry=1, customer.hash
retry #2	WARN	notify retry	retry=2, customer.hash
gave up	ERROR	notify failed	customer.hash, row_count
PII in backend	—	none	none (hashed only)

Rule of thumb. Reserve ERROR for terminal failures (retries that recover are WARN), keep bodies stable, and put only bounded/hashed attributes on records — never raw emails, tokens, or row contents. A log stream that cries wolf or leaks PII is worse than no logs at all.

Senior interview question on structured, correlated logging

A senior interviewer might ask: "Your pipeline logs are 40 GB/day of free-text print lines with no way to tie a message to a run, and an audit just found customer emails in them. Design the OpenTelemetry logging setup — the bridge, trace correlation, structured attributes, severity policy, and PII handling — and show how an on-call engineer goes from an alert to the exact failing task in minutes."

Solution Using the logging bridge, trace correlation, structured attributes, and PII redaction

import hashlib, logging
from opentelemetry import trace
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter
from opentelemetry.sdk.resources import Resource

# 1. One bridge for the whole process; resource carries stable identity.
provider = LoggerProvider(resource=Resource.create({
    "service.name": "airflow-worker", "pipeline.name": "daily_sales"}))
provider.add_log_record_processor(
    BatchLogRecordProcessor(OTLPLogExporter(endpoint="otel-collector:4317")))
logging.getLogger().addHandler(LoggingHandler(level=logging.INFO, logger_provider=provider))

# 2. A redacting filter enforces "no raw PII in attributes" centrally.
SENSITIVE = {"email", "token", "ssn", "password"}
def _hash(v): return hashlib.sha256(str(v).encode()).hexdigest()[:12]

class RedactFilter(logging.Filter):
    def filter(self, record):
        for k in list(getattr(record, "__dict__", {})):
            if k.lower() in SENSITIVE:
                setattr(record, k + "_hash", _hash(getattr(record, k)))
                delattr(record, k)                     # drop the raw value
        return True
logging.getLogger().addFilter(RedactFilter())

# 3. On-call path: alert fires on the failure Counter → open the trace → read logs.
tracer = trace.get_tracer("daily_sales")
log = logging.getLogger("daily_sales")

def task(name: str, table: str):
    with tracer.start_as_current_span(name) as span:
        span.set_attribute("target.table", table)
        for attempt in range(3):
            try:
                do_work(table)
                log.info("task ok", extra={"target.table": table})
                return
            except TransientError:
                log.warning("task retry", extra={"attempt": attempt + 1})
        log.error("task failed", extra={"target.table": table})   # terminal only
        span.set_status(trace.Status(trace.StatusCode.ERROR))
        raise

Step-by-step trace.

Layer	Component	Purpose
Bridge	LoggingHandler + OTLP exporter	stdlib logs → OTLP, one path
Identity	resource service.name/pipeline.name	who emitted, low cardinality
Correlation	log inside active span	trace_id/span_id stamped
Redaction	RedactFilter	raw PII → hashed attribute, centrally
Severity	WARN retry / ERROR terminal	trustworthy error rate
On-call	alert → trace → span_id logs	minutes to root cause

After the rollout, every message is a structured OTLP record carrying the run's trace_id; the RedactFilter guarantees no email/token value ever reaches the backend (only a 12-char hash); severities are honest so the error-rate alert means something; and the 40 GB of free text collapses into filterable, correlated events. On-call goes alert → open the trace by trace_id → filter logs by the red span's span_id → read the two lines that matter.

Output:

Metric	Before	After
Log correlation to run	none	every record carries trace_id
PII in backend	customer emails	12-char hashes only
Error-rate fidelity	inflated by retries	terminal failures only
Alert → failing task	~hours grepping	< 2 min via trace + span_id
Log searchability	free text	structured attributes

Why this works — concept by concept:

• Logging bridge — one LoggingHandler turns every existing stdlib log call into a correlated OTLP record, so adoption needs no call-site rewrite and legacy logs immediately gain trace_id/span_id.
• Log-inside-span correlation — because records are emitted while a span is active, the SDK stamps the run's trace_id/span_id, making the alert → trace → logs round-trip a few clicks instead of timestamp archaeology.
• Central RedactFilter — PII handling is enforced once at the handler, not trusted to every call site, so an interpolated email cannot leak; the hash preserves cross-event correlation without storing the value.
• Severity policy — WARN for recoverable retries, ERROR only for terminal failure, keeps the error rate honest so alerts are trusted and tail sampling keeps the right traces.
• Cost — structured records batch and compress far better than free text, and correlation removes the need to over-log "just in case." The eliminated cost is the 40 GB/day grep tax and the audit-failing PII exposure — O(1) lookup by trace_id versus O(GB) full-text scans.

Data validation
Topic — data-validation
Data validation problems on structured log auditing

Practice →

ETL Topic — etl ETL problems on failure handling and retries

Practice →

---

5. The OTLP Collector and instrumenting Airflow, Spark, and dbt

Instrument once with OTLP; the Collector decides where it all goes

The mental model in one line: the OTLP collector is a standalone service that receives telemetry over OTLP, runs it through a chain of processors (memory-limiter, batch, tail-sampling, resource/attribute editing), and exports it to one or more backends — deployed as a lightweight agent next to each workload and/or a central gateway — so your pipelines emit OTLP and nothing else, and every decision about sampling, redaction, batching, and destination lives in Collector config instead of in application code. Instrument Airflow, Spark, and dbt to speak OTLP; let the Collector own the routing.

Collector anatomy — receivers, processors, exporters, pipelines.

• Receivers. Ingest telemetry: otlp (gRPC 4317 / HTTP 4318) is the default; others exist (Prometheus scrape, filelog) but OTLP is the contract with your SDKs.
• Processors. Transform in order: memory_limiter (backpressure/OOM guard), batch (efficiency), tail_sampling (whole-trace decisions), resource/attributes (add/drop/redact), transform.
• Exporters. Send onward: otlp to a trace backend, prometheus for metrics scrape, otlphttp/vendor exporters for logs. One pipeline can have several exporters (fan-out).
• Pipelines. Under service.pipelines, wire traces, metrics, and logs each as receivers → processors → exporters. The three signals can share receivers but take different processor chains.

Agent vs gateway — the standard topology.

• Agent. A Collector per host/pod/node, close to the workload. It offloads batching and retry from the SDK, adds host metadata, and shields the app from backend outages.
• Gateway. A central Collector (a scaled deployment) that agents forward to. It owns expensive, global decisions — tail sampling (needs the whole trace), cross-source aggregation, and the single point where backends are configured.
• Why both. Agents give resilience and locality; the gateway gives one place to sample and route. Data pipelines especially want the gateway to make tail-sampling decisions across Airflow + Spark + dbt spans of the same trace.

Backpressure and reliability.

• memory_limiter first. It caps Collector memory and refuses/So throttles intake before an OOM, pushing backpressure to senders instead of crashing.
• Sending queue + retry. Exporters buffer and retry on backend blips, so a 30-minute backend outage does not drop telemetry — it drains when the backend returns.
• Tail sampling needs the gateway. Because a keep/drop decision depends on the entire trace (any span errored? total latency high?), it must run where all of a trace's spans converge — the gateway, not per-agent.

Instrumenting the three tools.

• Airflow (native). Set [traces] otel_on = True + otel_host/otel_port (2.10+) for DAG-run and task-instance spans, and [metrics] otel_on = True (2.7+) for scheduler/executor metrics. Add manual spans in tasks for sub-step detail and propagate traceparent via XCom.
• Spark (Java agent + listener). Attach -javaagent:opentelemetry-javaagent.jar to the driver and executors via spark.driver.extraJavaOptions / spark.executor.extraJavaOptions for JVM auto-instrumentation, and register a SparkListener to emit job/stage/task spans. Point the agent's OTLP exporter at the local Collector agent.
• dbt (wrapper). dbt emits no OTel, so wrap dbt build in a root span and, after the run, synthesise one span per model from run_results.json timings (section 2). Read the parent traceparent from an env var set by the Airflow task.

Common interview probes on the Collector.

• "Why run a Collector at all — can't the SDK export directly?" — it can, but the Collector centralises batching, retry, sampling, redaction, and backend choice; direct export couples every app to the backend.
• "Where does tail sampling run?" — the gateway, because the decision needs the whole trace.
• "How do you fan out to two backends?" — list two exporters on the pipeline.
• "How do you instrument Spark, which has no native OTel?" — Java agent + SparkListener, exporting to the local agent Collector.

Worked example — a Collector config with tail sampling and fan-out

Detailed explanation. The canonical gateway config: an OTLP receiver, a processor chain that guards memory, batches, and tail-samples, and multiple exporters so traces go to a trace backend, metrics to Prometheus, and logs to a log backend. Build it and read every block.

• Receiver. otlp on gRPC + HTTP.
• Processors. memory_limiter → tail_sampling → batch for traces; memory_limiter → batch for metrics/logs.
• Exporters. otlp/tempo, prometheus, otlphttp/loki — fan-out per pipeline.

Question. Write a Collector config that ingests OTLP, keeps all error/slow traces plus 5% baseline, and fans the three signals to three backends.

Input.

Block	Value
Receiver	otlp (4317/4318)
Trace processors	memory_limiter, tail_sampling, batch
Metric/log processors	memory_limiter, batch
Exporters	otlp/tempo, prometheus, otlphttp/loki

Code.

receivers:
  otlp:
    protocols:
      grpc: { endpoint: 0.0.0.0:4317 }
      http: { endpoint: 0.0.0.0:4318 }

processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 2000
    spike_limit_mib: 400
  batch:
    send_batch_size: 8192
    timeout: 5s
  tail_sampling:
    decision_wait: 30s          # wait for the whole trace before deciding
    policies:
      - name: keep-errors
        type: status_code
        status_code: { status_codes: [ERROR] }
      - name: keep-slow
        type: latency
        latency: { threshold_ms: 30000 }
      - name: sample-baseline
        type: probabilistic
        probabilistic: { sampling_percentage: 5 }

exporters:
  otlp/tempo:      { endpoint: tempo:4317, tls: { insecure: true } }
  prometheus:      { endpoint: 0.0.0.0:8889 }
  otlphttp/loki:   { endpoint: http://loki:3100/otlp }

service:
  pipelines:
    traces:
      receivers:  [otlp]
      processors: [memory_limiter, tail_sampling, batch]
      exporters:  [otlp/tempo]
    metrics:
      receivers:  [otlp]
      processors: [memory_limiter, batch]
      exporters:  [prometheus]
    logs:
      receivers:  [otlp]
      processors: [memory_limiter, batch]
      exporters:  [otlphttp/loki]

Step-by-step explanation.

1. The otlp receiver exposes gRPC (4317) and HTTP (4318) — the two endpoints every OTel SDK and the Java agent target. This is the single ingestion contract; sources need to know nothing else.
2. memory_limiter is listed first in every chain deliberately: it enforces the memory ceiling and applies backpressure before batching allocates, so a traffic spike throttles senders instead of OOM-killing the Collector.
3. tail_sampling sits only in the traces pipeline. decision_wait: 30s holds a trace's spans until they have all likely arrived, then applies policies: keep if any span is ERROR, keep if latency > 30 s, else keep 5%. Errors and slow traces are never sampled away.
4. Each pipeline lists its own exporter, so traces land in Tempo, metrics are exposed for Prometheus scrape, and logs go to Loki. To fan out a signal to two backends, you would simply list two exporters (e.g. exporters: [otlp/tempo, otlp/vendor]).
5. Swapping a backend is now a one-line edit here — change otlp/tempo to a vendor exporter — with zero changes to any pipeline's code. That is the decoupling that justifies running the Collector.

Output.

Signal	Pipeline path	Backend
Traces	otlp → memory_limiter → tail_sampling → batch	Tempo
Metrics	otlp → memory_limiter → batch	Prometheus
Logs	otlp → memory_limiter → batch	Loki
Error/slow traces	always kept	Tempo
Baseline traces	5% kept	Tempo

Rule of thumb. Put memory_limiter first, batch last, and tail_sampling only in the traces pipeline on the gateway. Fan out by listing multiple exporters, and treat the exporter block as the single place a backend migration touches.

Worked example — enabling native OTel in Airflow and the Spark Java agent

Detailed explanation. Two of the three tools are configured, not coded. Airflow 2.10+ emits DAG-run and task spans natively once you flip a config block; Spark auto-instruments the JVM when you attach the OpenTelemetry Java agent to driver and executors. Wire both to the local Collector agent.

• Airflow. airflow.cfg (or env) [traces] and [metrics] OTel blocks pointing at the Collector.
• Spark. -javaagent on driver + executors, with OTLP endpoint env vars.
• Endpoint. Both point at the node-local Collector agent (localhost:4317).

Question. Configure Airflow native OTel traces/metrics and attach the Spark OTel Java agent so both export to the local Collector agent.

Input.

Tool	Mechanism	Endpoint
Airflow	[traces]/[metrics] config	otel-agent:4318
Spark driver	-javaagent + env	localhost:4317
Spark executors	-javaagent + env	localhost:4317

Code.

# airflow.cfg — native OpenTelemetry (2.10+ for traces, 2.7+ for metrics)
[traces]
otel_on = True
otel_host = otel-agent
otel_port = 4318
otel_ssl_active = False

[metrics]
otel_on = True
otel_host = otel-agent
otel_port = 4318
otel_interval_milliseconds = 15000

# Spark submit — attach the OTel Java agent to driver AND executors
spark-submit \
  --conf "spark.driver.extraJavaOptions=-javaagent:/opt/otel/opentelemetry-javaagent.jar" \
  --conf "spark.executor.extraJavaOptions=-javaagent:/opt/otel/opentelemetry-javaagent.jar" \
  --conf "spark.driverEnv.OTEL_SERVICE_NAME=daily_transform" \
  --conf "spark.executorEnv.OTEL_SERVICE_NAME=daily_transform" \
  --conf "spark.driverEnv.OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317" \
  --conf "spark.executorEnv.OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317" \
  --conf "spark.driverEnv.OTEL_TRACES_EXPORTER=otlp" \
  --conf "spark.driverEnv.OTEL_METRICS_EXPORTER=otlp" \
  daily_transform.py

Step-by-step explanation.

1. The Airflow [traces] otel_on = True block makes the scheduler emit a span per DAG run and per task instance automatically — no task code needed for the skeleton trace. otel_host/otel_port point at the Collector agent's OTLP/HTTP endpoint.
2. [metrics] otel_on = True turns on Airflow's native OTel metrics (scheduler heartbeat, executor slots, task states) on a 15 s interval — instant scheduler SLIs. These complement the manual pipeline SLIs from section 3.
3. On Spark, the -javaagent on both spark.driver.extraJavaOptions and spark.executor.extraJavaOptions is the key: instrument the driver and every executor JVM, or executor-side work (the actual heavy lifting) emits nothing.
4. The OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 env sends spans/metrics to the node-local Collector agent, which handles batching/retry and forwards to the gateway. OTEL_SERVICE_NAME sets the resource identity so Spark telemetry is attributable.
5. Together, Airflow native OTel gives the scheduler-level trace and metrics for free, and the Spark agent gives JVM-level spans; the manual SparkListener and traceparent propagation (sections 2/5) stitch Spark under the Airflow trace so it is one waterfall, not two silos.

Output.

Source	Telemetry gained	Code required
Airflow traces	DAG-run + task-instance spans	config only
Airflow metrics	scheduler/executor SLIs	config only
Spark driver	JVM spans + metrics	-javaagent
Spark executors	executor JVM spans	-javaagent
Stitching	one trace across both	traceparent + listener

Rule of thumb. Get the free telemetry first: flip Airflow's native [traces]/[metrics] blocks and attach the Spark Java agent to both driver and executors, all pointing at the node-local Collector agent. Then add manual spans and traceparent propagation only where the auto-instrumentation leaves gaps.

Worked example — agent + gateway topology for cross-tool tail sampling

Detailed explanation. Tail sampling across Airflow, Spark, and dbt only works if every span of a trace reaches the same Collector before the keep/drop decision. That forces the agent + gateway topology: node-local agents forward to a central gateway, and the gateway alone does tail sampling. Lay out the two tiers and the routing.

• Agents. One per node; receive local OTLP, add host metadata, batch, forward to the gateway. No sampling.
• Gateway. Scaled central deployment; receives from all agents, runs tail_sampling (needs the whole trace), exports to backends.
• Consistency. All spans of one trace_id must land on the same gateway replica — use a load-balancing exporter keyed on trace_id.

Question. Design the agent and gateway configs so a trace spanning Airflow, Spark, and dbt is tail-sampled as a whole.

Input.

Tier	Role	Sampling
Agent (per node)	receive local, forward	none
Gateway (central)	aggregate, decide, export	tail_sampling
Routing	loadbalancing by trace_id	keeps a trace on one gateway

Code.

# --- Collector AGENT (runs on every node; forwards to the gateway) ---
receivers:
  otlp: { protocols: { grpc: { endpoint: 0.0.0.0:4317 }, http: { endpoint: 0.0.0.0:4318 } } }
processors:
  memory_limiter: { check_interval: 1s, limit_mib: 512 }
  resourcedetection: { detectors: [env, system] }   # add host metadata
  batch: {}
exporters:
  # Route by trace_id so ALL spans of a trace reach the SAME gateway replica
  loadbalancing:
    routing_key: traceID
    protocol: { otlp: { tls: { insecure: true } } }
    resolver: { dns: { hostname: otel-gateway, port: 4317 } }
service:
  pipelines:
    traces:  { receivers: [otlp], processors: [memory_limiter, resourcedetection, batch], exporters: [loadbalancing] }
    metrics: { receivers: [otlp], processors: [memory_limiter, batch], exporters: [loadbalancing] }
    logs:    { receivers: [otlp], processors: [memory_limiter, batch], exporters: [loadbalancing] }

# --- Collector GATEWAY (central; owns tail sampling + backends) ---
receivers:
  otlp: { protocols: { grpc: { endpoint: 0.0.0.0:4317 } } }
processors:
  memory_limiter: { check_interval: 1s, limit_mib: 4000 }
  tail_sampling:
    decision_wait: 30s
    policies:
      - { name: errors, type: status_code, status_code: { status_codes: [ERROR] } }
      - { name: slow,   type: latency,     latency: { threshold_ms: 30000 } }
      - { name: base,   type: probabilistic, probabilistic: { sampling_percentage: 5 } }
  batch: {}
exporters:
  otlp/tempo:    { endpoint: tempo:4317, tls: { insecure: true } }
  prometheus:    { endpoint: 0.0.0.0:8889 }
  otlphttp/loki: { endpoint: http://loki:3100/otlp }
service:
  pipelines:
    traces:  { receivers: [otlp], processors: [memory_limiter, tail_sampling, batch], exporters: [otlp/tempo] }
    metrics: { receivers: [otlp], processors: [memory_limiter, batch], exporters: [prometheus] }
    logs:    { receivers: [otlp], processors: [memory_limiter, batch], exporters: [otlphttp/loki] }

Step-by-step explanation.

1. Each agent receives OTLP from the workloads on its node (Airflow worker, Spark driver/executors, dbt wrapper), adds host metadata via resourcedetection, and forwards — it does no sampling, because it only ever sees a fragment of any trace.
2. The agent's loadbalancing exporter with routing_key: traceID is the linchpin: it hashes each span's trace_id to pick a gateway replica, guaranteeing every span of one trace — whether from Airflow, Spark, or dbt — lands on the same gateway. Without this, a trace's spans scatter across replicas and tail sampling sees partial traces.
3. The gateway receives from all agents and runs tail_sampling with decision_wait: 30s. Because it now holds the whole trace, "did any span error?" and "was total latency > 30 s?" are answerable — the decision is correct, not guessed from a fragment.
4. Only the gateway talks to backends, so backend config, credentials, and fan-out live in exactly one place. Agents are cheap and stateless; the gateway is the scaled, stateful tier.
5. This topology is why cross-tool tail sampling works at all: sampling that depends on the complete trace must run where the complete trace converges, and loadbalancing by trace_id is what makes it converge.

Output.

Concern	Agent tier	Gateway tier
Locality / host metadata	yes	—
Batching / backpressure	yes	yes
Route trace to one replica	loadbalancing by traceID	—
Whole-trace tail sampling	—	yes (decision_wait)
Backend config / fan-out	—	one place

Rule of thumb. Run agents for locality and resilience, a gateway for global decisions, and a loadbalancing exporter keyed on trace_id so every span of a trace reaches the same gateway. Tail sampling belongs only where the whole trace converges — never on the per-node agent.

Senior interview question on Collector topology and cross-tool instrumentation

A senior interviewer might ask: "Stand up OpenTelemetry for an Airflow → Spark → dbt platform end to end. Cover how you instrument each of the three tools, the Collector topology, where tail sampling runs and why, how you fan out to a trace/metrics/logs backend, and what happens to telemetry when the backend is down for 30 minutes."

Solution Using agent + gateway Collectors, native/agent/wrapper instrumentation, and buffered fan-out

# 1. Instrumentation per tool (config, not bespoke code where possible)
#    Airflow  : native [traces]/[metrics] otel_on=True → node agent :4318
#    Spark    : -javaagent on driver+executors, OTEL_EXPORTER_OTLP_ENDPOINT=localhost:4317
#    dbt      : wrapper root span + run_results.json → per-model spans (traceparent from env)

# 2. GATEWAY: tail sampling + buffered, retrying fan-out to three backends
processors:
  memory_limiter: { check_interval: 1s, limit_mib: 4000 }
  tail_sampling:
    decision_wait: 30s
    policies:
      - { name: errors, type: status_code, status_code: { status_codes: [ERROR] } }
      - { name: slow,   type: latency,     latency: { threshold_ms: 30000 } }
      - { name: base,   type: probabilistic, probabilistic: { sampling_percentage: 5 } }
  batch: {}
exporters:
  otlp/tempo:
    endpoint: tempo:4317
    retry_on_failure: { enabled: true, max_elapsed_time: 3600s }   # ride out outages
    sending_queue:    { enabled: true, queue_size: 100000 }        # buffer while down
  prometheus:    { endpoint: 0.0.0.0:8889 }
  otlphttp/loki: { endpoint: http://loki:3100/otlp }
service:
  pipelines:
    traces:  { receivers: [otlp], processors: [memory_limiter, tail_sampling, batch], exporters: [otlp/tempo] }
    metrics: { receivers: [otlp], processors: [memory_limiter, batch], exporters: [prometheus] }
    logs:    { receivers: [otlp], processors: [memory_limiter, batch], exporters: [otlphttp/loki] }

# 3. The dbt wrapper — the one tool with no auto-instrumentation
import json, os, subprocess
from datetime import datetime
from opentelemetry import trace
from opentelemetry.propagate import extract

tracer = trace.get_tracer("dbt")

def run_dbt_instrumented():
    parent = extract({"traceparent": os.environ["TRACEPARENT"]})     # from Airflow task
    with tracer.start_as_current_span("dbt_build", context=parent):
        subprocess.run(["dbt", "build", "--target", "prod"], check=False)
        for r in json.load(open("target/run_results.json"))["results"]:
            t = r["timing"][0]
            s = tracer.start_span(
                f"dbt.model.{r['unique_id']}",
                start_time=int(datetime.fromisoformat(
                    t["started_at"].replace("Z", "+00:00")).timestamp() * 1e9))
            s.set_attribute("dbt.status", r["status"])
            s.end(end_time=int(datetime.fromisoformat(
                t["completed_at"].replace("Z", "+00:00")).timestamp() * 1e9))

Step-by-step trace.

Layer	Component	Purpose
Airflow	native [traces]/[metrics]	DAG + task spans/metrics, no code
Spark	Java agent (driver+executors) + SparkListener	JVM + stage spans
dbt	wrapper span + run_results.json	per-model spans, back-dated
Agents	per node, loadbalancing by trace_id	route whole trace to one gateway
Gateway	tail_sampling + fan-out	whole-trace decision; 3 backends
Resilience	sending_queue + retry_on_failure	buffer + drain across outages

After the rollout, all three tools speak OTLP to their node-local agent; agents load-balance by trace_id so each run's Airflow, Spark, and dbt spans converge on one gateway replica; the gateway keeps every error/slow trace plus 5% baseline and fans traces/metrics/logs to Tempo/Prometheus/Loki. When Tempo is down for 30 minutes, the exporter's sending_queue buffers up to 100k items and retry_on_failure drains them when Tempo returns — no telemetry lost as long as the queue holds.

Output:

Metric	Value / behaviour
Tools emitting OTLP	Airflow (native), Spark (agent), dbt (wrapper)
Trace continuity	one trace per run across all three
Tail-sampling correctness	whole-trace (gateway + loadbalancing)
Backends	Tempo (traces), Prometheus (metrics), Loki (logs)
30-min backend outage	buffered + drained, no loss (within queue)
Backend swap effort	one exporter line on the gateway

Why this works — concept by concept:

• Config-first instrumentation — Airflow native OTel and the Spark Java agent give the bulk of the telemetry with zero bespoke code; only dbt, which has no emitter, needs a wrapper. Minimise custom code to what the ecosystem cannot auto-instrument.
• loadbalancing by trace_id — routing every span of a trace to one gateway replica is the precondition for whole-trace tail sampling across tools; without it the gateway sees fragments and samples wrongly.
• tail_sampling on the gateway — the keep/drop decision depends on the entire trace (any error? slow?), so it runs only where the trace converges, keeping 100% of the traces you actually debug while shedding boring baseline volume.
• sending_queue + retry_on_failure — the exporter buffers and retries, so a backend outage becomes a drain-later event rather than data loss, decoupling pipeline reliability from backend uptime.
• Cost — cheap stateless agents per node, a scaled gateway, ~1–5% overhead per instrumented process, and bounded retained volume via tail sampling. The eliminated cost is per-app backend coupling and the re-instrumentation tax on every vendor change — instrument once in OTLP, decide everything in Collector config.

Streaming
Topic — streaming
Streaming problems on telemetry pipelines and fan-out

Practice →

Design
Topic — design
Design problems on collector agent/gateway topology

Practice →

---

Cheat sheet — OpenTelemetry for data pipelines

• Signal → question map. Traces answer "where did the time go in this run" (span waterfall, critical path). Metrics answer "is the fleet healthy over time" (rated counters, histogram percentiles, sampled gauges). Logs answer "what exactly happened in this event" (structured, severity-tagged, correlated). Pick the signal by the question; debugging a trend with traces or a single run with metrics is the classic mis-hire tell.
• Span-per-task template. Wrap each unit of work in with tracer.start_as_current_span(name) as span:; set low-cardinality attributes (pipeline.name, source.table, spark.stage.id, dbt.model); nest child spans for expensive sub-steps; on failure span.record_exception(e) + set_status(Status(StatusCode.ERROR)) then re-raise. Prefer context managers so spans always end; optimise only spans on the critical path.
• traceparent propagation snippet. inject(carrier) on the way out of a task/process, extract(carrier) on the way in, then start_as_current_span(name, context=parent). Carriers: XCom (Airflow→Airflow), env TRACEPARENT (Airflow→Spark/dbt), Kafka record headers (producer→consumer). A dropped context = a fresh trace_id = an orphan trace; assert the traceparent key survives.
• Metric instrument selection. Counter for monotonic totals you rate (rows, bytes, failures); UpDownCounter for live levels you mutate (queue depth, in-flight); Histogram for distributions you need p50/p95/p99 from (durations, batch sizes) — tune the buckets to your real range; Observable (async) Gauge for sampled levels with no add moment (freshness lag, watermark age, disk).
• Temporality rule. Cumulative for Prometheus (it computes rate() itself); delta for OTLP-native/hosted backends and to survive restarts cleanly. Set preferred_temporality explicitly on the OTLP metric exporter — a mismatch makes counters appear to reset or deltas read as totals.
• Cardinality guardrails. Time-series count ≈ product of distinct attribute values. Stable identity (service.name, pipeline.name) → resource attributes, set once. Only bounded discriminators (stage, status, table, a small reason enum) → data-point attributes. Never label metrics with run_id, user_id, file_path; put per-item identity on spans. Fix an explosion centrally with a View allow-list, not by editing every call site.
• Log correlation handler. Attach the OTel LoggingHandler (logging bridge) to the root logger once; then log inside an active span so trace_id/span_id are stamped automatically. Stable body + structured extra={} attributes (never interpolated f-strings). A central redaction filter hashes/drop email/token/ssn; WARN for recoverable retries, ERROR only for terminal failure.
• Exemplars. Record a Histogram value inside an active, sampled span and enable exemplars, so a p99 bucket carries a trace_id — click the spike, land on a representative slow trace. The one-click metric→trace bridge a split Prometheus/ELK stack cannot give you.
• Collector pipeline skeleton. receivers: [otlp] → processors: (memory_limiter first, then signal-specific: tail_sampling on traces only, batch last) → exporters: (one per backend; list several to fan out). Wire service.pipelines.{traces,metrics,logs} each as receivers→processors→exporters. Backend swap = one exporter line.
• Agent + gateway topology. Node-local agents (locality, host metadata, batching, resilience, no sampling) forward via a loadbalancing exporter keyed on traceID to a scaled gateway; the gateway runs tail_sampling (decision_wait: 30s; keep all ERROR + slow, sample ~5% baseline) because whole-trace decisions need the whole trace to converge. Backends configured only on the gateway.
• Instrumentation matrix. Airflow: native [traces]/[metrics] otel_on = True (traces 2.10+, metrics 2.7+) + manual spans + XCom traceparent. Spark: -javaagent:opentelemetry-javaagent.jar on driver and executors + a SparkListener for stage spans. dbt: no native emitter — wrap dbt build in a root span and synthesise per-model spans from run_results.json timings (parent traceparent from env).
• Reliability. memory_limiter applies backpressure before OOM; exporter sending_queue + retry_on_failure buffer and drain across backend outages so a 30-minute blip is drain-later, not data loss. Tail sampling keeps 100% of the traces you actually debug (errors, slow) while shedding boring baseline volume — cost control without losing signal.

Frequently asked questions

What is OpenTelemetry in one sentence?

OpenTelemetry is a single vendor-neutral, CNCF-standard framework — an API, SDK, semantic conventions, and the OTLP wire protocol — for generating, collecting, and exporting the three telemetry signals (traces, metrics, and logs) from any system, so you instrument your code once and choose the observability backend as a downstream configuration detail rather than a code dependency. For a data pipeline that means one SDK stamps every Airflow task, Spark stage, and dbt model with correlated telemetry, and an OTLP Collector routes it to whatever backend — Tempo, Prometheus, Loki, or a commercial vendor — you point it at. The payoff is correlation and portability: all three signals share a trace_id, and swapping a backend never touches pipeline code.

Traces vs metrics vs logs — when do I use each?

Use traces to answer "where did the time go in this specific run" — a trace is a tree of spans, one per unit of work, and its waterfall exposes the critical path so you optimise the span that actually gates the run. Use metrics to answer "is the fleet healthy over time" — rated counters for throughput and failure rate, histograms for duration percentiles, async gauges for sampled levels like freshness lag. Use logs to answer "what exactly happened in this event" — structured, severity-tagged records that, because they are emitted inside an active span, carry the trace_id for one-click correlation. The classic mistake is debugging a single slow run from a metric dashboard (aggregates hide the individual) or computing a weekly trend by grepping logs (expensive and unbounded); match the signal to the question.

How does trace context propagate across Airflow, Spark, and dbt?

Through the W3C traceparent context, carried across every process boundary with the OTel propagation API: inject(carrier) serialises the current span's context on the way out, and extract(carrier) rebuilds it on the way in so the next span attaches as a child instead of starting a new trace_id. The carrier differs per hop — XCom between Airflow tasks, an environment variable (TRACEPARENT) passed to spark-submit and to the dbt wrapper, Kafka record headers between a producer and consumer — but the inject/extract discipline is identical. On the Spark side a SparkListener opens stage spans under the driver's extracted context; on the dbt side a wrapper reads traceparent from the env and back-dates one span per model from run_results.json. Skip propagation at any boundary and that one logical run fragments into several disconnected traces.

Do I need the OpenTelemetry Collector?

Technically no — SDKs can export OTLP straight to a backend — but in production you almost always want the Collector, because it is the layer that decouples your pipelines from everything operational. It centralises batching and retry (so a backend outage becomes a buffered drain-later event, not data loss), applies tail_sampling so you keep every error and slow trace while shedding boring baseline volume, edits and redacts attributes (dropping high-cardinality labels, hashing PII) before anything hits the wire, and owns the single place where backends are configured — so a vendor migration is one exporter line, not a re-instrumentation of every host. The standard topology is a lightweight agent per node (locality, resilience) forwarding to a central gateway (whole-trace tail sampling, fan-out). Direct SDK export couples every application to the backend and scatters sampling decisions; the Collector is what makes "instrument once" real.

OpenTelemetry vs Prometheus + ELK — is it a replacement?

Not a replacement so much as a unification and a decoupling. A split Prometheus (metrics) + ELK (logs) stack has no first-class tracing and — critically — no shared identifier, so a latency spike in Prometheus and an error in Kibana can only be correlated by eyeballing timestamps across two UIs. OpenTelemetry adds native traces, gives all three signals a shared trace_id so an exemplar jumps from a metric bucket to the exact trace and a log carries the same trace_id, and turns the backend into swappable Collector config. And it does not force you to abandon your existing tools: Prometheus and Elasticsearch become exporters behind the Collector, so you keep the backends you like and gain tracing plus one-click cross-signal correlation on top. Frame it as "instrument once, correlate everything, keep the backends optional."

Delta vs cumulative temporality — which do I pick?

Match it to your backend. Cumulative temporality reports the running total since process start on every export; Prometheus expects this and computes rates itself with rate(). Delta temporality reports only what changed since the previous export; many OTLP-native and hosted backends prefer it, and it survives process restarts cleanly because a restart does not look like a counter reset. The failure mode is a mismatch — sending cumulative to a delta-expecting backend (or vice versa) yields counters that appear to reset or deltas that read as totals, so your rates are nonsense. Set preferred_temporality explicitly on the OTLP metric exporter rather than relying on a default, and if you fan out to two backends with different expectations, run two metric pipelines with the appropriate temporality each.

Practice on PipeCode

• Drill the ETL practice library → for the pipeline-orchestration, failure-handling, and lineage problems that OpenTelemetry instrumentation makes debuggable.
• Rehearse pipeline internals on the data processing practice library → for the Spark job/stage and transformation scenarios where spans and the critical path earn their keep.
• Sharpen the streaming axis with the streaming practice library → for the telemetry fan-out, backpressure, and Collector-style routing patterns.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the three-signal model and the traceparent-propagation decision against real graded inputs — traces, metrics, logs, and the Collector.

Lock in OpenTelemetry muscle memory

Docs explain the signals. PipeCode drills explain the decision — when a trace answers a question a metric can't, when `traceparent` has to cross the Airflow → Spark → dbt boundary, when an ID-valued metric label becomes a runaway bill, and when tail sampling has to move to the gateway. Pipecode.ai is Leetcode for Data Engineering — observability-first practice tuned for the production trade-offs senior data engineers actually face.

Practice ETL problems →
Practice data processing problems →