Structured Logging for Data Pipelines: JSON Logs & Correlation IDs

structured logging is the difference between a 3 a.m. incident where you grep a gigabyte of plaintext for the word "error" and pray, and one where you type a single query — filter by run, filter by task, group by error class — and watch the failing record surface in under a second. A data pipeline emits logs from dozens of tasks, retries, workers, and subprocesses, and every one of those lines is either an opaque string a human has to eyeball or a machine-readable event a query engine can slice. The distinction is not cosmetic. It decides whether your on-call rotation can answer "which of last night's 40,000 orders failed validation, and why?" with a query, or whether that question takes an engineer an afternoon of awk and regret.

This guide is the walkthrough you wished existed the first time an incident review asked "why couldn't we tell which run this log line belonged to?" It moves in layers: why a log line should be an event and not a string, how to shape those events as json logs under a stable schema so every field is query-able, how to thread a correlation id (and its cousin the trace id) through every hop so a whole run's logs join on one value, how log levels, sampling, and PII redaction keep volume and risk bounded, and finally how to ship those events into a log aggregation platform where dashboards and alerts fall out of the structured fields for free. 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. Examples are Python and PostgreSQL-flavoured, but the model carries to any language and any pipeline orchestrator.

When you want hands-on reps immediately after reading, drill the log-processing practice library →, rehearse on the etl practice library →, and sharpen your parsing intuition on the JSON practice library →.

---

On this page

• Why structured logging beats grep
• JSON logs and a consistent log schema
• Correlation and trace IDs across tasks
• Log levels, sampling, and PII redaction
• Shipping and querying logs
• Cheat sheet — structured logging recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why structured logging beats grep

A log line is either a string you grep or an event you query — structured logging picks the second

The one-sentence invariant: structured logging is the practice of emitting every log line as a machine-parseable event — a set of typed key-value fields (usually a JSON object) with a stable schema — rather than a human-readable sentence, so that downstream tooling can filter, group, aggregate, and alert on individual fields instead of running substring searches over free text. The moment a pipeline has more than one task, more than one worker, and more than one run in flight, free-text logs stop scaling: the questions you need to answer at 3 a.m. ("which run? which task? which record? what error class? how many?") are field questions, and free text has no fields. Structured logging front-loads a small amount of discipline — decide your keys, emit objects, never f"processed {n} rows" again — and in exchange every question becomes a query.

The four axes that decide whether logs are useful.

• Parseability. Can a machine extract fields without a fragile regex? Free text forces a per-format grok pattern that breaks the day someone reorders a sentence. A JSON object is parsed once, universally, and every key becomes a column.
• Correlation. Can you reassemble one logical unit of work — one run, one request, one record — from log lines scattered across tasks and hosts? This requires a shared join key (a correlation id) present on every line. Free text almost never carries one consistently.
• Cardinality and cost. Every field you log has a cost in index size and query speed. High-cardinality fields (per-record ids, raw payloads) are powerful for debugging and expensive to index. Structured logging makes the cost visible and chooseable; free text hides it until the bill arrives.
• Retention and PII. Logs are retained, indexed, and searched by many people. A raw email address or token in a log line is a compliance incident waiting for an audit. Structured fields can be redacted by key deterministically; free text has to be scrubbed with heuristics that always miss one.

The 2026 reality — JSON on stdout, aggregation platform behind it.

• JSON to stdout is the default. The Twelve-Factor convention — the process writes structured events to stdout/stderr, the platform captures and routes them — is now near-universal for containerised pipelines. The application does not manage log files, rotation, or shipping; it emits objects and the collector does the rest.
• An aggregation platform does the indexing. Loki, Elasticsearch/OpenSearch, Datadog, CloudWatch Logs, Splunk — the specific product varies, but the shape is identical: a collector parses the JSON, an index makes fields query-able, and a dashboard queries them. Structured logging is the contract that makes all of these work well.
• Plaintext still survives — in local dev. A human tailing a terminal wants a readable line, not a wall of JSON. The mature setup emits JSON in production and pretty-prints the same structured event for a human in development. The event is structured either way; only the renderer differs.
• Logs, metrics, and traces converge. Structured log fields feed metrics (count by event, sum rows) and correlate with traces (shared trace_id). Getting the log schema right is what lets one signal reinforce the other two.

What interviewers listen for.

• Do you say "a log line is an event, not a string" unprompted? — senior signal.
• Do you name a correlation id as the join key for reassembling a run before being asked? — required answer.
• Do you talk about cardinality cost — that high-cardinality fields are expensive to index — rather than "log everything"? — senior signal.
• Do you separate schema (keys) from rendering (JSON vs pretty) — the same event, two renderers? — senior signal.
• Do you flag PII redaction as a first-class concern, not an afterthought? — required answer.

Worked example — the plaintext-to-event transform

Detailed explanation. The single most useful mental exercise is to take one real, bad plaintext log line and rewrite it as a structured event. Everything about structured logging follows from doing this transform deliberately: you name the invariant part (the event), lift the variable parts into typed fields, and attach the context (run, task, correlation id) that free text always drops.

• The bad line. "2026-09-05 02:14:07 INFO processed 1240 rows for orders in 3.1s" — timestamp, level, and a sentence with three numbers baked into English.
• The problem. To count rows per table you must regex for (\w+) and processed (\d+); the day someone writes "loaded" instead of "processed" the regex silently returns zero.
• The fix. Emit {"ts": ..., "level": "info", "event": "rows_processed", "table": "orders", "rows": 1240, "duration_ms": 3100} — now sum(rows) by table is a query, not a regex.

Question. Convert the plaintext line into a structured event and list the queries the structured form unlocks that the plaintext form cannot answer cheaply.

Input.

Piece of the line	Plaintext form	Structured field
when	2026-09-05 02:14:07	ts (ISO-8601 / epoch)
severity	INFO	level = info
what happened	processed ... rows	event = rows_processed
which table	for orders	table = orders
how many	1240 rows	rows = 1240
how long	in 3.1s	duration_ms = 3100

Code.

import json, time

# BAD: a sentence. Every number is trapped inside English.
def log_bad(table, n, secs):
    print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} INFO processed {n} rows for {table} in {secs}s")

# GOOD: an event. Every number is a typed, query-able field.
def log_event(table, n, secs):
    print(json.dumps({
        "ts": time.time(),            # epoch seconds; the collector adds ISO too
        "level": "info",
        "event": "rows_processed",    # the stable, low-cardinality name
        "table": table,               # a dimension you will group by
        "rows": n,                    # a measure you will sum
        "duration_ms": round(secs * 1000),
    }))

log_bad("orders", 1240, 3.1)
log_event("orders", 1240, 3.1)

Step-by-step explanation.

1. The plaintext line packs six facts into one string; five of them are only recoverable by parsing English. Any change to the wording — a synonym, a reordering, an extra clause — breaks whichever regex a dashboard relied on, and it breaks silently, returning zero matches rather than an error.
2. The structured line names the invariant part once as event = "rows_processed". This is the low-cardinality name of the thing that happened; it never contains the variable data. Everything variable becomes its own key.
3. table is a dimension — a low-to-medium cardinality field you group and filter by. rows and duration_ms are measures — numbers you sum, average, and percentile. Separating dimensions from measures up front is the whole game.
4. Types matter: rows is an integer, not the string "1240 rows". A query engine can sum() an integer field; it cannot sum a substring. Emitting the right type is what turns a log field into an aggregate.
5. The same event object can be rendered as pretty text for a human in dev and as compact JSON for the collector in prod. The event is the contract; the renderer is a detail.

Output.

Question	Plaintext answer	Structured answer
Rows per table today	regex + manual sum	sum(rows) by table
p95 task duration	not feasible	quantile(duration_ms, 0.95)
Error rate by event	grep "error", eyeball	count by (event) where level="error"
Which run emitted this	usually unknowable	filter run_id (once added)

Rule of thumb. Never bake a number or an identifier into an English sentence. Name the event once, lift every variable into a typed field, and let the query engine do the arithmetic. If you find yourself about to write a regex against your own logs, you logged them wrong.

Worked example — scoring a pipeline's logging on the four axes

Detailed explanation. Before you redesign logging, score the current state on the four axes — parseability, correlation, cardinality cost, and PII/retention — so the redesign targets the real gap. Most pipelines are strong on one axis and blind on the others; naming which one is missing is what makes the fix cheap.

• Parseability score. Are lines JSON (10), semi-structured key=value (5), or free text (0)?
• Correlation score. Is there a run/request id on every line (10), on some (5), or none (0)?
• Cardinality discipline. Are high-cardinality fields deliberate and bounded (10), unbounded but present (5), or is everything dumped including raw payloads (0)?
• PII/retention. Are PII fields redacted by policy (10), redacted ad hoc (5), or logged raw (0)?

Question. Score a legacy Airflow pipeline that prints free-text lines with a timestamp and level but no run id, dumps full row payloads on error, and logs customer emails. Recommend the highest-leverage single change.

Input.

Axis	Observation	Score
Parseability	free text, logging default format	0
Correlation	no run id on lines	0
Cardinality	full row payloads on error	0
PII	emails logged raw	0

Code.

# A tiny scorer you can run against a sample of real log lines.
import json, re

def score_line(line: str) -> dict:
    parseable = 10 if _is_json(line) else (5 if "=" in line and " " in line else 0)
    has_corr  = 10 if re.search(r'(run_id|correlation_id|trace_id)', line) else 0
    pii       = 0 if re.search(r'[\w.]+@[\w.]+', line) else 10   # naive email probe
    return {"parseable": parseable, "correlation": has_corr, "pii_safe": pii}

def _is_json(line: str) -> bool:
    try:
        obj = json.loads(line)
        return isinstance(obj, dict)
    except ValueError:
        return False

sample = '2026-09-05 02:14:07 INFO processed 1240 rows for [e-mail]'
print(score_line(sample))   # {'parseable': 0, 'correlation': 0, 'pii_safe': 0}

Step-by-step explanation.

1. The scorer parses a sample of production lines, not the whole stream — a few hundred lines are enough to characterise the format. _is_json is the parseability probe: if a line loads as a dict, it is structured.
2. The correlation probe looks for any of the three join keys (run_id, correlation_id, trace_id). Zero hits means logs from different runs are indistinguishable once interleaved — the single most damaging gap.
3. The PII probe is deliberately naive (an email regex). Its job is not to catch every case but to prove the point: raw PII is present, so redaction must be built before retention grows.
4. All four axes scoring zero is the common legacy starting point. The instinct is to fix everything at once; the correct move is to sequence the changes by leverage.
5. The highest-leverage single change is adding a correlation id/run_id to every line — even before converting to JSON. A run id on free text still lets you grep one run out of the interleaved stream, which is the difference between a solvable and an unsolvable incident. JSON conversion comes next; redaction ships with it.

Output.

Change	Effort	Leverage	Order
Add run_id to every line	low	high (isolate a run)	1
Convert to JSON events	medium	high (query-able)	2
Redact PII by key	medium	high (compliance)	3 (with 2)
Bound high-cardinality fields	low	medium (cost)	4

Rule of thumb. Score before you refactor. If exactly one axis is missing, fix that one; if all four are missing, add a correlation id first (it rescues even free text), then convert to JSON with redaction, then bound cardinality. Sequence by leverage, not by ease.

Data engineering interview question on structured logging fundamentals

A senior interviewer often opens with: "Our pipeline logs are free-text lines printed by 40 tasks across 12 workers, with no run identifier. Last night one of 40,000 orders failed validation and we spent three hours finding it. Walk me through how you'd convert this to structured logging, what the first change would be, and how the same incident would go next time."

Solution Using structured JSON events keyed by a run id

# structured_logging.py — the minimal upgrade that fixes the incident class
import json, logging, sys, time, uuid, contextvars

# 1. A context variable carries the run id without threading it through every call.
run_id_var: contextvars.ContextVar[str] = contextvars.ContextVar("run_id", default="-")

class JsonFormatter(logging.Formatter):
    """Render every record as one JSON object with a stable key set."""
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": round(record.created, 3),
            "level": record.levelname.lower(),
            "event": record.getMessage(),         # the low-cardinality name
            "run_id": run_id_var.get(),            # the join key on EVERY line
            "logger": record.name,
        }
        # Structured extras passed via logger.info("evt", extra={"fields": {...}})
        fields = getattr(record, "fields", None)
        if fields:
            payload["fields"] = fields
        return json.dumps(payload)

def get_logger() -> logging.Logger:
    log = logging.getLogger("pipeline")
    if not log.handlers:
        h = logging.StreamHandler(sys.stdout)
        h.setFormatter(JsonFormatter())
        log.addHandler(h)
        log.setLevel(logging.INFO)
    return log

# 2. Set the run id once at the start of the run; every line inherits it.
def run_pipeline(orders):
    run_id_var.set(uuid.uuid4().hex[:12])
    log = get_logger()
    log.info("run_started", extra={"fields": {"n_orders": len(orders)}})
    failed = 0
    for o in orders:
        if not validate(o):
            failed += 1
            log.error("order_validation_failed",
                      extra={"fields": {"order_id": o["id"], "reason": o["reason"]}})
    log.info("run_finished", extra={"fields": {"failed": failed}})

def validate(o):  return o.get("reason") is None

Step-by-step trace.

Step	Before (free text, no run id)	After (JSON events + run_id)
Line format	English sentence	one JSON object per event
Run isolation	impossible (interleaved)	filter run_id = "a1b2c3d4e5f6"
Find the failed order	3 h of grep/awk	event="order_validation_failed" → fields.order_id
Count failures	manual	count where event=... and run_id=...
Reason for failure	not captured	fields.reason
Human-readable in dev	yes	yes (pretty renderer over same event)

After the change, the incident becomes a two-line query: filter the run id, filter event = "order_validation_failed", and read fields.order_id and fields.reason. The failing order surfaces in under a second, with its reason attached, instead of after three hours of substring archaeology.

Output:

Metric	Before	After
Time to isolate one run	not possible	one filter
Time to find failed order	~3 h	< 1 s
Failure reason available	no	yes (fields.reason)
Lines parseable by machine	0%	100%
Extra code	—	one formatter + one contextvar

Why this works — concept by concept:

• Event, not string — event is the stable, low-cardinality name of what happened (order_validation_failed), never the variable data. Variable data lives in fields. This split is what makes count by event and filter by event trivially fast.
• Run id as the join key — a single run_id on every line reassembles one logical run from lines scattered across 40 tasks and 12 workers. Without it, interleaved logs are noise; with it, one filter isolates the run.
• contextvars carry context — the run_id is set once per run and read by the formatter, so no function has to thread it through its signature. contextvars are async- and thread-safe, so concurrent runs don't leak each other's ids.
• extra fields, one schema — passing extra={"fields": {...}} keeps the top-level keys fixed (ts, level, event, run_id) while allowing per-event detail in a nested bag. Queries against reserved keys stay stable as events evolve.
• Cost — one JSON serialise per line (O(fields) — microseconds) and one contextvar read (O(1)). The eliminated cost is O(hours) of human search per incident and O(lines) of regex maintenance. Net: a few microseconds per log call buys constant-time incident triage.

Logs
Topic — log-processing
Log-processing and event-parsing problems

Practice →

Data Topic — etl ETL problems on pipeline events

Practice →

---

2. JSON logs and a consistent log schema

One event, one JSON object, a stable key set — the schema is the contract your queries depend on

The mental model in one line: json logs are the pattern where every log line is a single JSON object with a small set of reserved top-level keys (ts, level, event, run_id, and a nested fields bag) that never change, so that every downstream query, dashboard, and alert can rely on those keys existing with those types — the schema is a contract, and breaking it silently breaks every consumer that joined on it. The value of structured logging is entirely proportional to how consistent the structure is: a hundred services each inventing their own key for "the thing that happened" is nearly as useless as free text, because no cross-service query can span them. The discipline is to fix a tiny reserved vocabulary and push everything else into a namespaced fields object.

The reserved top-level keys — the vocabulary every event shares.

• ts. The event time. Emit epoch seconds (or RFC3339) from a single clock authority. Let the collector add an ingestion timestamp separately — never conflate event time and ingest time, or late-arriving logs will lie about when things happened.
• level. The severity — debug, info, warn, error, critical. Lowercased, from a fixed enum. This drives routing, sampling, and alert thresholds, so it must be a controlled value, never free text.
• event. The stable, low-cardinality name of what happened (rows_processed, order_validation_failed, run_started). This is the field you group by; it must never contain variable data like ids or counts.
• run_id / correlation_id. The join key that reassembles one logical unit of work. Present on every line without exception. (Section 3 goes deep on how it propagates.)
• fields. A nested object holding everything event-specific — order_id, rows, duration_ms, reason. Namespacing the variable data keeps the top-level schema fixed while events evolve freely.

Types and cardinality — the two decisions that make or break query speed.

• Types are load-bearing. A number logged as a number can be summed; logged as a string it cannot. Booleans, integers, floats, and timestamps must keep their JSON types. The most common structured-logging bug is stringifying everything.
• Cardinality is a budget. event and level are low-cardinality (dozens of values) and cheap to index. order_id and trace_id are high-cardinality (millions of values) and expensive. Put low-cardinality fields where they get indexed as labels; keep high-cardinality fields in the JSON body where they're searchable but not indexed as dimensions.
• Reserved keys are sacred. Never let an event-specific field collide with a reserved key. If an event has its own notion of "level" or "event", it goes in fields under a different name. Collisions silently overwrite the contract.
• Nesting depth is bounded. Keep fields one level deep where possible. Deeply nested objects are painful to query in most log platforms (dotted-path access, exploded mappings). Flatten to fields.order_id, not fields.order.metadata.id.

Rendering — one event, two audiences.

• Production renderer. Compact single-line JSON to stdout. No pretty-printing (multi-line JSON breaks line-oriented collectors). One object, one line, one newline.
• Development renderer. The same structured event, pretty-printed as HH:MM:SS LEVEL event key=value key=value for a human tailing a terminal. The event is identical; only the formatter differs, chosen by an env var.
• Never two code paths. The anti-pattern is a separate logger.info(f"...") for humans and logger.info(event, extra=...) for machines. There is one call site; the renderer decides the surface form.

Common interview probes on JSON log schemas.

• "What are your reserved keys?" — ts, level, event, run_id/correlation_id, fields.
• "Where do event-specific fields go?" — a namespaced fields bag, never at the top level.
• "How do you evolve the schema without breaking queries?" — add keys, never rename or retype; deprecate with overlap.
• "Why not deeply nest?" — query ergonomics and index explosion; keep fields shallow.

Worked example — a structlog processor chain that enforces the schema

Detailed explanation. The cleanest way to guarantee every event carries the reserved keys is a processor chain — an ordered list of functions each event passes through before rendering. structlog in Python is the canonical implementation, but the pattern (merge context → add timestamp → add level → render JSON) is language-agnostic. Build the chain that stamps ts, level, run_id, and renders JSON.

• Processor 1. Merge context-local bindings (the run_id set for this run).
• Processor 2. Add an ISO timestamp under ts.
• Processor 3. Add the level name.
• Processor 4. Render the event dict as compact JSON (prod) or pretty (dev).

Question. Configure a structlog processor chain that emits the reserved-key schema and lets callers add event-specific fields as keyword arguments.

Input.

Processor	Responsibility
merge_contextvars	inject run_id bound for this run
add_timestamp	add ts from one clock
add_log_level	add level from the call
JSONRenderer / ConsoleRenderer	prod vs dev surface form

Code.

# logging_setup.py — structlog processor chain enforcing the reserved schema
import os, structlog, logging, sys

def configure_logging():
    is_prod = os.getenv("ENV", "dev") == "prod"

    renderer = (structlog.processors.JSONRenderer()      # compact one-line JSON
                if is_prod
                else structlog.dev.ConsoleRenderer())    # pretty for humans

    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,     # inject run_id, etc.
            structlog.processors.add_log_level,          # -> "level"
            structlog.processors.TimeStamper(fmt="iso", key="ts"),  # -> "ts"
            structlog.processors.EventRenamer("event"),  # rename "event" key explicitly
            renderer,
        ],
        wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        logger_factory=structlog.PrintLoggerFactory(file=sys.stdout),
        cache_logger_on_first_use=True,
    )

# Usage — the caller names the event and passes fields as kwargs
configure_logging()
log = structlog.get_logger()

structlog.contextvars.bind_contextvars(run_id="a1b2c3d4e5f6")
log.info("rows_processed", table="orders", rows=1240, duration_ms=3100)
# prod  -> {"table":"orders","rows":1240,"duration_ms":3100,"run_id":"a1b2c3d4e5f6","level":"info","ts":"2026-09-05T02:14:07Z","event":"rows_processed"}
# dev   -> 2026-09-05T02:14:07Z [info     ] rows_processed  run_id=a1b2c3d4e5f6 table=orders rows=1240 duration_ms=3100

Step-by-step explanation.

1. merge_contextvars runs first so that context-local bindings (the run_id set once per run via bind_contextvars) are merged into every event dict before anything else touches it. This is how the join key lands on every line without being passed to each call.
2. add_log_level stamps level from the method used (log.info → info), guaranteeing the reserved key is always a controlled enum value, never a free-text string a caller might mistype.
3. TimeStamper(fmt="iso", key="ts") adds the event time under the reserved key ts from one clock, so every event across every process is comparable. Event time is stamped at emit; the collector adds ingest time separately downstream.
4. EventRenamer("event") makes the positional message argument land under the reserved key event — the low-cardinality name. Callers pass table=..., rows=... as keyword arguments, which become top-level fields in this flat example (nest them under fields if your platform prefers).
5. The renderer is chosen by ENV: JSONRenderer for prod (one compact line the collector parses) and ConsoleRenderer for dev (pretty, aligned, human-readable). Same event dict, two surface forms, one call site — the anti-pattern of two logging code paths is structurally impossible.

Output.

Environment	Renderer	Surface form
prod	JSONRenderer	{"event":"rows_processed","table":"orders",...}
dev	ConsoleRenderer	... [info] rows_processed run_id=... table=orders ...
both	same processors	reserved keys ts,level,event,run_id always present

Rule of thumb. Enforce the reserved schema in a processor chain, not in each call site. Callers name the event and pass fields as keyword arguments; the chain guarantees ts, level, and run_id exist on every line. One chain, one contract, zero drift.

Worked example — evolving the schema without breaking queries

Detailed explanation. Schemas evolve: a new field is needed, an old one is deprecated, a type turns out wrong. The rule that keeps a fleet of dashboards alive is additive-only evolution — you may add keys and add enum values, but you may never rename a key, retype a value, or repurpose a name, because every one of those silently breaks a query that joined on the old shape. Walk through evolving an event from duration_s (seconds, float) to duration_ms (milliseconds, int) safely.

• The wrong way. Rename duration_s → duration_ms and change the unit. Every dashboard querying duration_s now returns null; every alert threshold set in seconds now fires on milliseconds.
• The right way. Add duration_ms alongside duration_s, dual-write both for a deprecation window, migrate dashboards, then drop duration_s after the retention window has aged out old data.

Question. Migrate a duration field from seconds-as-float to milliseconds-as-int without breaking existing dashboards, and describe the deprecation timeline.

Input.

Phase	duration_s	duration_ms	Dashboards read
before	present (float)	absent	duration_s
overlap	present (float)	present (int)	migrating
after	absent	present (int)	duration_ms

Code.

# Additive evolution: dual-write during the deprecation window.
def log_task_done(log, table, secs):
    ms = round(secs * 1000)
    log.info(
        "task_done",
        table=table,
        duration_s=round(secs, 3),   # DEPRECATED: kept for the overlap window
        duration_ms=ms,              # NEW: canonical going forward
    )

# A schema guard in CI: fail the build if a reserved key or type changes.
RESERVED = {"ts": str, "level": str, "event": str, "run_id": str}

def assert_schema(sample_event: dict):
    for key, typ in RESERVED.items():
        assert key in sample_event, f"missing reserved key: {key}"
        assert isinstance(sample_event[key], typ), f"{key} wrong type"
    # New fields are allowed (additive); removals of reserved keys are not.

Step-by-step explanation.

1. During the overlap phase the producer dual-writes both duration_s (deprecated) and duration_ms (canonical). No consumer breaks, because the old key still exists with its old type and unit.
2. Dashboards and alerts migrate one at a time from duration_s to duration_ms, converting thresholds from seconds to milliseconds as they go. Because both fields are present, a half-migrated fleet is fully functional.
3. The assert_schema guard runs in CI against a captured sample event. It enforces that reserved keys exist with the right types — a rename or retype of a reserved key fails the build, catching the breaking change before it ships.
4. New fields are explicitly allowed by the guard: additive evolution is safe, so the guard only forbids removing or retyping reserved keys, not adding new ones. This encodes the additive-only rule as an automated check.
5. Only after the log retention window has fully aged out (say 30 days, so no dashboard queries old data expecting duration_s) is the deprecated field removed from the producer. The timeline is: add → overlap ≥ retention window → migrate consumers → drop.

Output.

Change kind	Safe?	Why
Add a new field	yes	additive; old queries unaffected
Add an enum value to level	yes	additive; existing filters still match
Rename a key	no	old queries return null silently
Change a field's type	no	aggregates and thresholds break
Drop a field after overlap	yes	only once no query reads it

Rule of thumb. Treat the log schema like a public API: additive-only, never rename or retype, deprecate with an overlap window at least as long as your log retention. Enforce the reserved keys with a CI schema guard so a breaking change fails the build, not the 3 a.m. dashboard.

Data engineering interview question on log schema design

A senior interviewer might ask: "Design a company-wide JSON log schema for a platform of 60 pipelines written in Python, Java, and Go. It must let a single query span all of them, keep index cost bounded, and evolve without breaking dashboards. Specify the reserved keys, where event-specific data goes, how you handle types and cardinality, and how you enforce the schema across three languages."

Solution Using a reserved-key schema with a namespaced fields bag and a shared contract

// The reserved-key contract — identical across Python, Java, Go.
// Every log line is ONE object with exactly these top-level keys.
{
  "ts":       "2026-09-05T02:14:07.123Z",   // RFC3339, one clock authority
  "level":    "info",                         // enum: debug|info|warn|error|critical
  "event":    "rows_processed",               // low-cardinality name; NEVER variable data
  "service":  "orders-etl",                   // which pipeline emitted it
  "run_id":   "a1b2c3d4e5f6",                 // the join key; present on EVERY line
  "trace_id": "0af7651916cd43dd8448eb211c80319c",  // W3C trace id when present
  "fields": {                                  // everything event-specific, namespaced
    "table":       "orders",
    "rows":        1240,
    "duration_ms": 3100
  }
}

# shared contract expressed once; each language binds to it.
# Python reference implementation used as the CI fixture for all three langs.
import json, time, contextvars

_ctx = contextvars.ContextVar("logctx", default={})

RESERVED = ("ts", "level", "event", "service", "run_id", "trace_id", "fields")

def emit(level: str, event: str, **fields):
    ctx = _ctx.get()
    line = {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime()) + "Z",
        "level": level,
        "event": event,                       # stays low-cardinality
        "service": ctx.get("service", "unknown"),
        "run_id": ctx.get("run_id", "-"),
        "trace_id": ctx.get("trace_id", ""),
        "fields": fields,                     # all variable data lands here
    }
    print(json.dumps(line, separators=(",", ":")))   # compact, one line

def bind(**kv):
    _ctx.set({**_ctx.get(), **kv})

-- The payoff: one query spans all 60 pipelines because keys are shared.
-- (Expressed as SQL over a parsed logs table; the same shape maps to LogQL/OpenSearch.)
SELECT service,
       fields ->> 'table'          AS table_name,
       COUNT(*)                     AS error_events,
       MAX(ts)                      AS last_seen
FROM   logs
WHERE  level = 'error'
  AND  ts >= now() - INTERVAL '1 hour'
GROUP  BY service, fields ->> 'table'
ORDER  BY error_events DESC;

Step-by-step trace.

Concern	Decision	Reasoning
Reserved keys	ts, level, event, service, run_id, trace_id, fields	fixed vocabulary every language emits
Event-specific data	nested fields object	top-level stays stable as events evolve
Cross-language enforcement	shared JSON fixture in CI	Python/Java/Go each assert against it
Cardinality	event/level/service low; ids in body	index labels cheap, ids searchable
Types	ints stay ints in fields	aggregates work; no stringification
Cross-pipeline query	group by service, fields->>'table'	one query spans all 60 pipelines

After adoption, a single query filters level = 'error' across every pipeline and groups by service and table, because all 60 pipelines share the reserved keys. A new pipeline in any of the three languages is query-able on day one by conforming to the same fixture; index cost stays bounded because only the low-cardinality keys are indexed as labels.

Output:

Metric	Value
Reserved top-level keys	7 (fixed)
Languages conforming	Python, Java, Go
Cross-pipeline query	one, spans all 60
Indexed dimensions	service, event, level (low-cardinality)
High-cardinality ids	in body, searchable, not indexed as labels
Schema drift caught	in CI, against shared fixture

Why this works — concept by concept:

• Reserved keys as a shared vocabulary — fixing ts, level, event, service, run_id, trace_id means a query written once spans every pipeline. The value of structured logging scales with structural consistency, and a shared vocabulary is what makes 60 pipelines queryable as one.
• Namespaced fields bag — pushing event-specific data into fields keeps the top-level schema fixed while events evolve. Adding a new measure never touches the reserved keys, so no cross-pipeline query breaks.
• Cardinality-aware placement — low-cardinality dimensions (service, event, level) are indexed as labels; high-cardinality ids live in the searchable body. This bounds index size and cost while keeping ids findable.
• Contract enforced in CI — a shared JSON fixture that each language asserts against turns "please follow the schema" into an automated gate. Drift fails the build, not the dashboard.
• Cost — O(fields) to serialise each line and O(1) index writes for the low-cardinality labels; query cost is O(matching lines) over an index rather than O(all lines) over free text. The eliminated cost is the combinatorial explosion of per-service ad-hoc formats. Net: constant per-line overhead buys cross-fleet query-ability.

JSON
Topic — json
JSON parsing and schema problems

Practice →

Logs Topic — log-processing Log-processing problems on structured events

Practice →

---

3. Correlation and trace IDs across tasks

One id generated at the edge, propagated through every hop — the thread that stitches a whole run's logs together

The mental model in one line: a correlation id is a single opaque identifier generated once at the entry point of a logical unit of work — a pipeline run, an API request, a message — and propagated unchanged into every task, retry, subprocess, queue message, and downstream call, so that filtering logs by that one value returns every line the work produced, no matter how many services or hosts it touched. Structured logging gives you fields; the correlation id gives you the join. Without it, a well-structured log fleet is still a pile of disconnected events; with it, any run collapses to a single filter. Its cousin the trace_id does the same job for distributed tracing, and the mature setup carries both on the same lines.

correlation_id vs trace_id vs span_id — three ids, three scopes.

• correlation_id. The business/run-level join key. One per logical run. It is yours — you generate it, you name it, you decide its format. It exists to answer "show me everything about this run."
• trace_id. The distributed-tracing root id (W3C Trace Context, 16 bytes / 32 hex chars). One per distributed trace, shared by every span. Tracing tools (OpenTelemetry, Jaeger, Tempo) understand it natively. Log it alongside the correlation id so logs and traces cross-link.
• span_id. The per-operation id (8 bytes / 16 hex) inside a trace. Identifies one step (one task, one call). Logs can carry the current span_id so a log line pins to an exact span.
• traceparent. The W3C header (version-traceid-spanid-flags, e.g. 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01) that carries trace_id + parent span_id + sampling flag across process boundaries. This is the wire format for propagation.

Context propagation — how the id survives each boundary.

• In-process (async/threads). contextvars bind the id to the current logical context; child tasks inherit it automatically. This is the clean way — no parameter threading, no globals that leak across concurrent runs.
• Across orchestrator tasks (Airflow, Dagster, Prefect). The run id is the natural correlation id; propagate it through the DAG run_id, XCom, or task parameters so every task binds the same value. Never let each task generate its own.
• Across process/subprocess. Pass the id as an environment variable or CLI argument to the child, which binds it on startup. The child's logs then join the parent's run.
• Across queues (Kafka, SQS, Pub/Sub). Put the id in a message header, not the payload, so infrastructure and consumers can read it without deserialising the body. The consumer binds it before processing.
• Across HTTP/gRPC. Carry traceparent (and optionally a custom X-Correlation-Id) as a request header. The server extracts and binds it. This is where W3C Trace Context earns its keep — it is the standard every framework understands.

Generation and format rules.

• Generate once, at the true edge. The id is created where work first enters the system (the scheduler that starts the run, the API gateway that accepts the request). Everything downstream propagates, never regenerates.
• Accept an inbound id if present. If a caller already supplied a traceparent or correlation id, adopt it rather than minting a new one — that is what stitches your logs to the caller's.
• Use a compact, collision-resistant format. A UUID4 hex or a W3C 32-hex trace id. Long enough to avoid collisions, short enough to eyeball.
• Never reuse across runs. A correlation id that repeats makes two runs indistinguishable — the exact failure the id exists to prevent.

Common interview probes on correlation.

• "Where is the correlation id generated?" — once, at the edge; propagated everywhere else.
• "How does it cross a Kafka boundary?" — in a message header, bound by the consumer.
• "What's the difference between correlation_id and trace_id?" — business/run scope vs distributed-trace scope; carry both.
• "How do you avoid leaking ids across concurrent runs?" — contextvars, not globals.

Worked example — a contextvar-bound logger that stamps the correlation id

Detailed explanation. The foundation of propagation is binding the id to a context variable so every log call in that context inherits it without being passed the id explicitly. Build a small middleware that generates (or adopts) a correlation id at the start of a unit of work, binds it, and guarantees every subsequent log line carries it — even across await points and thread-pool hops.

• Generate or adopt. Read an inbound traceparent/correlation id; mint a UUID4 if absent.
• Bind. Set it on a contextvars.ContextVar.
• Stamp. The formatter reads the contextvar for every line.
• Reset. Clear the binding when the unit of work ends so it never leaks to the next.

Question. Implement a correlation_scope context manager that binds a correlation id for the duration of a unit of work and guarantees every log line inside it carries the id.

Input.

Step	Behaviour
enter scope	adopt inbound id or generate UUID4
inside scope	every log line carries correlation_id
nested awaits/threads	inherit the same id
exit scope	reset binding (no leak)

Code.

import contextvars, contextlib, uuid, json, time

correlation_id_var: contextvars.ContextVar[str] = contextvars.ContextVar(
    "correlation_id", default="-"
)

@contextlib.contextmanager
def correlation_scope(inbound: str | None = None):
    """Bind a correlation id for the duration of a unit of work."""
    cid = inbound or uuid.uuid4().hex        # adopt if given, else mint
    token = correlation_id_var.set(cid)      # bind on the contextvar
    try:
        yield cid
    finally:
        correlation_id_var.reset(token)      # reset so it never leaks

def log(event: str, **fields):
    print(json.dumps({
        "ts": round(time.time(), 3),
        "level": "info",
        "event": event,
        "correlation_id": correlation_id_var.get(),   # stamped automatically
        "fields": fields,
    }, separators=(",", ":")))

# One unit of work; every line inside shares the id.
with correlation_scope() as cid:
    log("run_started")
    log("rows_processed", table="orders", rows=1240)
    log("run_finished")
# -> all three lines carry the same "correlation_id"

Step-by-step explanation.

1. correlation_scope accepts an optional inbound id. If a caller (an upstream service, a queue message header) already supplied one, we adopt it — this is what links our logs to the caller's. If not, we mint a fresh UUID4. The id is generated exactly once per scope.
2. correlation_id_var.set(cid) binds the id to the context variable and returns a token. Because it is a contextvars.ContextVar, the binding is local to the current logical context and is inherited by child tasks spawned inside the scope — including across await boundaries and ThreadPoolExecutor submissions that copy the context.
3. Every log call reads correlation_id_var.get(), so the id lands on every line automatically. No function signature carries the id; no global is mutated. This is the difference between clean propagation and threading an argument through fifty functions.
4. The finally: reset(token) restores the previous binding when the scope exits. This is critical in long-lived workers that process many units of work on the same thread — without the reset, run N+1 would inherit run N's id and the two runs would be indistinguishable.
5. The result is the invariant the correlation id exists to provide: filter by one value, get every line the unit of work produced, and nothing from any other unit.

Output.

Log line	correlation_id
run_started	f3a1...
rows_processed	f3a1... (same)
run_finished	f3a1... (same)
next scope's first line	9c72... (new, no leak)

Rule of thumb. Bind the correlation id to a contextvar inside a scope with a guaranteed reset, generate-or-adopt at the edge, and let the formatter stamp it. Never pass the id as a function argument, and never store it in a plain global — that is how concurrent runs bleed into each other.

Worked example — propagating the id across a Kafka boundary

Detailed explanation. The hardest propagation boundary is asynchronous messaging: the producer and consumer are different processes, possibly different services, separated in time. The id must ride in the message header (not the body), so the consumer can bind it before deserialising or processing the payload. Build the producer that injects and the consumer that extracts.

• Producer. On send, read the current correlation id from the contextvar and set it as a Kafka header (bytes).
• Consumer. On receive, read the header, open a correlation_scope(inbound=header), and process inside it.
• Result. Producer-side and consumer-side logs join on the same id across the async gap.

Question. Wire correlation-id propagation through Kafka so a message produced in run A is processed under run A's id in the consumer.

Input.

Side	Action	Where the id lives
producer	inject id on send	Kafka record header correlation_id
consumer	extract + bind before processing	correlation_scope(inbound=...)
logs	both sides carry same id	joined across the queue

Code.

# Producer — inject the current correlation id as a message header.
def produce(producer, topic, key, value):
    cid = correlation_id_var.get()
    headers = [("correlation_id", cid.encode("utf-8"))]   # header, NOT body
    producer.produce(topic, key=key, value=value, headers=headers)
    log("message_produced", topic=topic, key=key)

# Consumer — extract the header and process inside a matching scope.
def consume_loop(consumer):
    while True:
        msg = consumer.poll(1.0)
        if msg is None or msg.error():
            continue
        inbound = _header(msg, "correlation_id")          # may be None
        with correlation_scope(inbound=inbound):
            log("message_received", topic=msg.topic())
            process(msg.value())                          # its logs carry the id too
            log("message_processed")

def _header(msg, name):
    for k, v in (msg.headers() or []):
        if k == name:
            return v.decode("utf-8")
    return None

Step-by-step explanation.

1. The producer reads the current correlation id from the contextvar — the id bound by whatever correlation_scope the producing code runs inside — and encodes it as a Kafka header. Headers are metadata alongside the record; putting the id there (not in the value) means consumers and infrastructure can read it without deserialising the payload schema.
2. The consumer polls a message and extracts the correlation_id header via _header. If the producer set it, the consumer adopts it; if not (a message from a legacy producer), inbound is None and the scope mints a fresh id, degrading gracefully.
3. The consumer opens correlation_scope(inbound=inbound) before processing. Every log line emitted inside — including deep inside process(msg.value()) — inherits the producer's id through the contextvar, exactly as if producer and consumer shared a call stack.
4. The join now spans the async gap: a query for the correlation id returns the producer's message_produced line and the consumer's message_received / message_processed lines, even though they happened in different processes minutes apart.
5. This same pattern generalises: SQS message attributes, Pub/Sub attributes, HTTP traceparent headers, gRPC metadata. The boundary changes; the rule ("id in the header, bind on receive") does not.

Output.

Event	Process	correlation_id
message_produced	producer (run A)	f3a1...
message_received	consumer	f3a1... (from header)
message_processed	consumer	f3a1... (from scope)
downstream call logs	consumer	f3a1... (inherited)

Rule of thumb. Propagate the correlation id in message headers and request headers, never in the body, and bind it on the receiving side before any processing runs. Adopt an inbound id when present; mint one only at the true edge. The header is the wire; the contextvar is the in-process carrier.

Data engineering interview question on correlation across tasks

A senior interviewer might ask: "You have an Airflow DAG of 12 tasks that fan out to Kafka and back, running on a pool of workers. Currently each task's logs are impossible to correlate — you can't tell which run a log line belongs to. Design correlation-id propagation across the whole DAG, including the fan-out to Kafka and a retried task, so that one query returns every line for one DAG run."

Solution Using the DAG run_id as the correlation id propagated through contextvars and message headers

# 1. Bind the DAG run_id as the correlation id at the start of every task.
from airflow.decorators import task
import contextvars, uuid, json, time

correlation_id_var = contextvars.ContextVar("correlation_id", default="-")

def bind_correlation(context) -> str:
    # The Airflow run_id is stable across all tasks and retries of one DAG run.
    cid = context["run_id"]                    # e.g. "manual__2026-09-05T02:14:00"
    correlation_id_var.set(cid)
    return cid

def log(event, **fields):
    print(json.dumps({
        "ts": round(time.time(), 3), "level": "info", "event": event,
        "correlation_id": correlation_id_var.get(),
        "task": fields.pop("task", None), "fields": fields,
    }, separators=(",", ":")))

# 2. Every task binds the SAME id from the shared DAG context.
@task
def ingest(**context):
    cid = bind_correlation(context)
    log("task_started", task="ingest")
    rows = read_source()
    # 3. Fan out to Kafka WITH the id in the header (survives the async hop).
    for r in rows:
        produce("orders.raw", key=r["id"], value=r, headers=[("correlation_id", cid.encode())])
    log("task_finished", task="ingest", rows=len(rows))

@task(retries=3)
def transform(**context):
    bind_correlation(context)                  # retry re-binds the SAME run_id
    log("task_started", task="transform")
    ...                                         # its logs still carry the run's id
    log("task_finished", task="transform")

-- 4. The payoff query: every line for one DAG run, across all 12 tasks + retries.
SELECT ts, task, event, fields
FROM   logs
WHERE  correlation_id = 'manual__2026-09-05T02:14:00'
ORDER  BY ts ASC;

Step-by-step trace.

Hop	Mechanism	id carried
DAG start	Airflow generates run_id	canonical correlation id
task → task	each task binds context["run_id"]	same id, all 12 tasks
task → Kafka	id in message header	survives async hop
Kafka → consumer	consumer binds header id	id crosses services
task retry	re-bind run_id on each attempt	retries share the id
query	filter correlation_id = run_id	whole run in one result

After the change, the DAG run_id becomes the correlation id every task binds — including on retries, because Airflow keeps the same run_id across attempts. The fan-out to Kafka carries the id in headers, so the consumer's logs join too. One query on correlation_id returns every line from all 12 tasks, their retries, and the Kafka consumers, ordered by time.

Output:

Metric	Before	After
Lines correlatable to a run	~0%	100%
Retry logs joined to run	no	yes (same run_id)
Kafka consumer logs joined	no	yes (header id)
Query to see a whole run	not possible	one filter
New id minted per task	yes (chaotic)	no (one per run)

Why this works — concept by concept:

• run_id as the natural correlation id — the orchestrator already mints one stable identifier per run and keeps it across retries. Reusing it as the correlation id means zero new id-generation logic and guarantees retries stay joined to their run.
• contextvar binding per task — each task binds the shared run_id on entry, so every log call in the task inherits it without argument threading. Concurrent DAG runs on the same worker stay isolated because contextvars are context-local.
• header propagation across Kafka — putting the id in the message header carries it across the async, cross-process boundary, so consumer logs join the producing run's logs on the same value.
• retry re-binding — because retries reuse the run_id, re-binding on each attempt keeps all attempts under one correlation id; you see the failed attempt and the successful retry in one query.
• Cost — O(1) to bind per task and O(1) to attach a header per message; the query is O(matching lines) over an indexed correlation_id. The eliminated cost is the impossibility of run isolation — an unbounded human cost per incident. Net: constant per-hop overhead buys whole-run reconstruction with a single filter.

Logs
Topic — log-processing
Log-processing problems on correlation and joins

Practice →

Data Topic — etl ETL problems on multi-task pipelines

Practice →

---

4. Log levels, sampling, and PII redaction

Levels route and cost, sampling caps volume, redaction keeps PII out of the index — three dials on the same stream

The mental model in one line: log levels, sampling, and redaction are the three control dials on log volume and safety — the level is a routing and cost signal (which lines to keep, ship, and alert on), sampling caps the volume of high-frequency events without losing the rare important ones, and redaction guarantees that no PII field ever reaches an index that dozens of people can search — and a pipeline that gets any one of the three wrong either drowns in cost, misses the signal, or ships a compliance incident. Structured logging makes all three deterministic: levels are a controlled enum, sampling keys off structured fields, and redaction targets fields by name rather than scrubbing free text with hopeful regexes.

The levels and what each means in a pipeline.

• debug. Developer-only detail — the payload of a record, the SQL about to run, the branch taken. Off in production by default; enabled dynamically for a service or run when debugging. High volume, low retention.
• info. The normal narrative — run_started, rows_processed, run_finished. The backbone of dashboards and throughput metrics. Kept, but the highest-frequency info events are the sampling candidates.
• warn. Something recoverable happened — a retry fired, a record was quarantined, a fallback path was taken. Not an incident yet, but a leading indicator. Never sampled away.
• error. A unit of work failed — a task, a record, a call. Always kept, always shippable, usually alertable. This is the level on-call watches.
• critical. The pipeline (or a component) is down or corrupting data. Pages immediately. Rare by design; if critical is noisy, it has been misused.

Sampling — capping volume without losing the signal.

• Head-based sampling. Decide at emit time whether to keep a line (e.g. keep 1% of a chatty debug/info event). Cheap and simple; the risk is dropping the one sampled-out line that would have explained an incident.
• Tail-based sampling. Buffer a unit of work's lines and decide after you know the outcome — keep 100% of runs that errored, sample 1% of runs that succeeded. More expensive (requires buffering) but keeps every line of every failure. This is the pattern for correlated logs and traces.
• Key the sample on structured fields. Sample by event (thin the chattiest events only) or by correlation_id hash (keep or drop a whole run coherently, so a kept run has all its lines). Never sample randomly per line — that shreds runs into unreadable fragments.
• Always keep errors and warnings. Sampling applies to high-frequency low-severity events. warn/error/critical are never sampled away.

PII redaction — keeping regulated data out of the index.

• Redact by key, not by regex. Structured fields let you name the sensitive keys (email, ssn, card_number, phone) and redact them deterministically. This beats scanning free text, which always misses a format.
• Allow-list beats deny-list. The safest schema declares which fields are allowed in logs; everything else is dropped or masked. A deny-list (redact these known-bad keys) fails the moment a new PII field is added upstream.
• Mask, hash, or drop — choose per field. Mask for partial visibility (a***@example.com), hash when you still need to join or count distinct without the raw value (sha256(email)), drop entirely when the field has no logging value.
• Redact before shipping, at the earliest point. Redaction is a processor in the chain that runs before the line leaves the process. PII must never hit the collector, the index, or disk — redacting downstream is too late; it has already been persisted.

Common interview probes on levels, sampling, and PII.

• "What's the difference between head-based and tail-based sampling?" — decide at emit vs after outcome; tail keeps every failure.
• "How do you sample without breaking correlated logs?" — key the sample on correlation_id so whole runs are kept or dropped.
• "How do you keep PII out of logs?" — redact by key with an allow-list, mask/hash/drop, before shipping.
• "Why not just DEBUG everything and filter later?" — volume cost, PII exposure, and index blowup; levels are a cost dial, not a mood.

Worked example — level-based routing to different sinks

Detailed explanation. Levels are most powerful when they route: info goes to the cheap high-volume index, error/critical also fan out to the alerting sink, and debug is dropped in prod unless dynamically enabled. Build a handler that routes by level so cost and alerting are controlled by the enum, not by ad-hoc if statements at every call site.

• debug. Dropped in prod; enabled per-service via an env/flag.
• info/warn. To the primary index sink.
• error/critical. To the primary index and the alert sink.

Question. Implement level-based routing where error and above are duplicated to an alert sink while debug is suppressed in production.

Input.

Level	Primary index	Alert sink	Dropped in prod
debug	no	no	yes (unless enabled)
info	yes	no	no
warn	yes	no	no
error	yes	yes	no
critical	yes	yes	no

Code.

import os, json, sys, time

LEVELS = {"debug": 10, "info": 20, "warn": 30, "error": 40, "critical": 50}
MIN_LEVEL = LEVELS[os.getenv("LOG_LEVEL", "info")]     # dynamic floor
ALERT_FLOOR = LEVELS["error"]

def emit(level: str, event: str, **fields):
    lvl = LEVELS[level]
    if lvl < MIN_LEVEL:
        return                                          # suppressed (e.g. debug in prod)
    line = json.dumps({
        "ts": round(time.time(), 3), "level": level,
        "event": event, "correlation_id": correlation_id_var.get(),
        "fields": fields,
    }, separators=(",", ":"))
    sys.stdout.write(line + "\n")                       # primary index (collector reads stdout)
    if lvl >= ALERT_FLOOR:
        _to_alert_sink(line)                            # duplicate error+ to alerting

def _to_alert_sink(line: str):
    # e.g. write to a dedicated stream/topic the alerting pipeline consumes
    sys.stderr.write(line + "\n")

Step-by-step explanation.

1. MIN_LEVEL is read from LOG_LEVEL at startup, so the floor is a deployment-time (and, with a reload hook, run-time) dial. Setting LOG_LEVEL=debug for one service turns its debug stream on without a code change; the default info suppresses debug in prod.
2. Any line below MIN_LEVEL returns immediately — it is never serialised, never written, never shipped. This is where the cost dial lives: dropping debug at the source saves serialisation, network, and index cost, not just display.
3. Lines at or above the floor are serialised once and written to stdout, which the collector reads as the primary index feed. One serialisation, reused for both sinks.
4. Lines at or above ALERT_FLOOR (error) are also written to the alert sink. Routing is driven purely by the level enum — no call site needs an if is_error: also_alert(). The severity decides the fan-out.
5. Because the level is a controlled enum stamped by the logging chain (not free text), the routing is deterministic: a mistyped "eror" can't slip a real error past the alert floor, because levels come from the fixed set.

Output.

Emitted level	Written to stdout	Written to alert sink
debug (prod)	no	no
info	yes	no
warn	yes	no
error	yes	yes
critical	yes	yes

Rule of thumb. Route by level in one place, not with if statements at every call site. Make the level floor a deployment dial so you can raise verbosity for one service or run without shipping code, and always duplicate error+ to the alert sink from the routing layer.

Worked example — correlation-coherent tail sampling

Detailed explanation. Random per-line sampling destroys correlated logs: keeping 1% of lines leaves you with fragments of thousands of runs and complete records of none. The fix is to sample by correlation_id so a run is kept or dropped as a whole, and to keep 100% of runs that errored. Build a sampler that hashes the correlation id for a coherent head-sample and force-keeps any run that emitted an error.

• Coherent head-sample. hash(correlation_id) % 100 < sample_pct keeps whole runs.
• Force-keep on error. If any line in a run is error+, keep the whole run (tail decision).
• Result. Every kept run is complete; every failed run is complete.

Question. Implement sampling that keeps 1% of successful runs (coherently) and 100% of runs containing any error.

Input.

Rule	Decision basis
sample rate	1% of runs
coherence	key on correlation_id hash
force-keep	any error+ in the run
never sampled	warn/error/critical lines

Code.

import hashlib, json

SAMPLE_PCT = 1              # keep 1% of successful runs

def _run_is_sampled(correlation_id: str) -> bool:
    h = int(hashlib.sha256(correlation_id.encode()).hexdigest(), 16)
    return (h % 100) < SAMPLE_PCT           # whole run kept-or-dropped coherently

def buffer_and_decide(run_lines: list[dict]) -> list[dict]:
    """Tail decision: called when a unit of work finishes."""
    cid = run_lines[0]["correlation_id"]
    errored = any(l["level"] in ("error", "critical") for l in run_lines)
    if errored:
        return run_lines                     # keep 100% of failed runs, all lines
    if _run_is_sampled(cid):
        return run_lines                     # keep this sampled successful run, all lines
    # Drop the run's info/debug, but NEVER drop its warnings.
    return [l for l in run_lines if l["level"] in ("warn", "error", "critical")]

Step-by-step explanation.

1. _run_is_sampled hashes the correlation_id and keeps the run if the hash falls in the first SAMPLE_PCT percent of the space. Because the decision keys on the run id, every line of a kept run is kept and every line of a dropped run is dropped — the run stays coherent, never fragmented.
2. buffer_and_decide is a tail decision: it runs when the unit of work finishes, so it knows the outcome. errored checks whether any line reached error/critical.
3. If the run errored, every line is kept regardless of the sample — you want the full narrative of every failure, including the debug/info lines that explain it. This is the core advantage of tail sampling over head sampling.
4. If the run succeeded and was sampled in, all its lines are kept as a representative healthy trace. If it succeeded and was sampled out, the info/debug lines are dropped — but warn/error/critical are always retained, so leading-indicator warnings are never lost to sampling.
5. The cost trade is explicit: tail sampling requires buffering a run's lines until it completes (memory proportional to concurrent runs × lines per run), in exchange for complete records of every failure and every sampled success. For high-volume pipelines this is the right trade; head sampling is the cheaper fallback when buffering is infeasible.

Output.

Run outcome	Sampled in?	Lines kept
errored	n/a	100% (all levels)
success, sampled in	yes	100% (all levels)
success, sampled out	no	warn/error/critical only
success, sampled out, no warns	no	0 (info/debug dropped)

Rule of thumb. Sample by correlation_id so runs stay whole, decide at the tail so every failure is complete, and never let sampling drop a warn or above. Random per-line sampling is an anti-pattern — it optimises volume by destroying the exact correlation that makes structured logs useful.

Worked example — a PII redaction processor with an allow-list

Detailed explanation. Redaction must be a processor that runs before the line leaves the process, and the safe design is an allow-list: declare which field names may appear in logs; mask, hash, or drop everything else. Build a redactor that masks known PII, hashes join-critical PII, and drops unknown fields, so a raw email can never reach the index even if a developer logs it by accident.

• Mask. email → a***@example.com (partial visibility for debugging).
• Hash. user_id when it's PII-adjacent → sha256 (still joinable, not reversible).
• Drop. Anything not on the allow-list → removed entirely.

Question. Implement a redaction processor that masks emails, hashes user ids, and drops any field not on the allow-list, then show it stopping an accidental raw-email log.

Input.

Field	Policy
email	mask (a***@domain)
user_id	hash (sha256, 12 hex)
rows, table, duration_ms	allow (safe)
anything else	drop

Code.

import hashlib, re

ALLOWED = {"rows", "table", "duration_ms", "status", "reason"}
MASK    = {"email", "phone"}
HASH    = {"user_id", "account_id"}

def _mask_email(v: str) -> str:
    m = re.match(r"(.).*(@.*)", v)
    return f"{m.group(1)}***{m.group(2)}" if m else "***"

def _hash(v) -> str:
    return hashlib.sha256(str(v).encode()).hexdigest()[:12]

def redact_fields(fields: dict) -> dict:
    out = {}
    for k, v in fields.items():
        if k in MASK:
            out[k] = _mask_email(v) if k in ("email",) else "***"
        elif k in HASH:
            out[k] = _hash(v)                 # joinable, not reversible
        elif k in ALLOWED:
            out[k] = v                        # safe field, passes through
        else:
            continue                          # UNKNOWN -> dropped (allow-list)
    return out

# Accidental raw-email log is neutralised before it can ship.
raw = {"email": "[e-mail]", "user_id": 4412, "rows": 1240, "notes": "VIP customer"}
print(redact_fields(raw))
# -> {'email': 'a***@example.com', 'user_id': 'e3b0c44298fc', 'rows': 1240}
#    ('notes' dropped: not on the allow-list)

Step-by-step explanation.

1. The processor classifies each field by name against three sets: MASK, HASH, and ALLOWED. Everything else falls through to the else: continue — it is dropped. This is the allow-list stance: unknown fields are removed by default, so a new PII field added upstream can never leak.
2. email is masked to a***@example.com, preserving just enough for a human to recognise the domain and first letter during debugging without exposing the full address to everyone with index access.
3. user_id is hashed with SHA-256. The hash is deterministic, so you can still count distinct users or join two log lines about the same user, but the raw id is not recoverable from the log. This is the pattern for PII-adjacent fields you need to correlate but not reveal.
4. rows, table, duration_ms pass through untouched — they carry no personal data and are the measures dashboards depend on. The allow-list makes "safe" an explicit, reviewed decision.
5. The accidental notes: "VIP customer" field — exactly the kind of free-text field that leaks PII — is dropped because it is not on the allow-list. A deny-list would have missed it; the allow-list catches it by construction. The processor runs in the chain before the line is serialised to stdout, so the raw value never touches the collector, index, or disk.

Output.

Input field	Output	Policy applied
email: [e-mail]	a***@example.com	mask
user_id: 4412	e3b0c44298fc	hash
rows: 1240	1240	allow
notes: "VIP customer"	(dropped)	allow-list default

Rule of thumb. Redact by key with an allow-list, run the redactor in the logging chain before the line ships, and choose mask/hash/drop per field. Never rely on a deny-list or free-text scrubbing — the allow-list is the only design that stays safe when someone adds a new field upstream without telling you.

Data engineering interview question on level, sampling, and PII policy

A senior interviewer might ask: "A high-volume ingestion pipeline emits 500,000 log lines per minute, including customer emails in some debug lines, and your log bill has tripled. Design a policy covering log levels, sampling, and PII redaction that cuts volume and cost, keeps every failure fully logged, and guarantees no PII reaches the index. Include how you'd enable deep debugging for one run without turning debug on globally."

Solution Using a level floor, correlation-coherent tail sampling, and an allow-list redactor

# 1. Level floor as a dynamic dial; debug off globally, on per-run.
import os, json, hashlib

LEVELS = {"debug":10,"info":20,"warn":30,"error":40,"critical":50}

def min_level(correlation_id: str) -> int:
    # Global floor is INFO; a targeted run can be boosted to DEBUG via a control table.
    if correlation_id in DEBUG_RUNS:          # set for ONE run when investigating
        return LEVELS["debug"]
    return LEVELS[os.getenv("LOG_LEVEL", "info")]

# 2. Redaction processor (allow-list) runs BEFORE anything ships.
ALLOWED, MASK, HASH = {"rows","table","duration_ms","status","reason"}, {"email","phone"}, {"user_id"}

def redact(fields: dict) -> dict:
    out = {}
    for k, v in fields.items():
        if   k in MASK:    out[k] = "***"
        elif k in HASH:    out[k] = hashlib.sha256(str(v).encode()).hexdigest()[:12]
        elif k in ALLOWED: out[k] = v
    return out                                 # unknown keys dropped

# 3. Correlation-coherent tail sampling at run completion.
SAMPLE_PCT = 1
def keep_run(run_lines: list[dict]) -> list[dict]:
    cid = run_lines[0]["correlation_id"]
    if any(l["level"] in ("error","critical") for l in run_lines):
        return run_lines                                     # 100% of failures
    if int(hashlib.sha256(cid.encode()).hexdigest(),16) % 100 < SAMPLE_PCT:
        return run_lines                                     # 1% of successes, whole
    return [l for l in run_lines if l["level"] in ("warn","error","critical")]

# 4. The emit path ties them together.
def emit(level, event, correlation_id, **fields):
    if LEVELS[level] < min_level(correlation_id):
        return None
    return {"level": level, "event": event,
            "correlation_id": correlation_id, "fields": redact(fields)}

-- 5. Control table: boost ONE run to debug without a global change.
INSERT INTO debug_runs(correlation_id, expires_at)
VALUES ('manual__2026-09-05T02:14:00', now() + INTERVAL '2 hours');

Step-by-step trace.

Control	Mechanism	Effect on volume / safety
Level floor	global INFO; drop DEBUG at source	cuts the largest volume bucket
Per-run debug	debug_runs control table	deep debug for one run only
Redaction	allow-list processor pre-ship	no PII in index, by construction
Tail sampling	keep 1% success runs, 100% failures	volume down, every failure intact
Warns retained	never sampled away	leading indicators preserved

After the policy ships, DEBUG is dropped at the source globally (the largest volume bucket disappears), a single run can be boosted to DEBUG via the control table for a two-hour investigation window, tail sampling keeps 1% of successful runs coherently plus 100% of failed runs, and the allow-list redactor guarantees emails and other PII never reach the index. Volume and cost fall sharply while every failure remains fully logged.

Output:

Metric	Before	After
Lines/min shipped	500,000	~40,000 (est.)
PII in index	present	none (allow-list)
Failure completeness	partial (sampled)	100%
Deep-debug one run	global toggle	per-run control row
Warnings retained	inconsistent	always

Why this works — concept by concept:

• Level as a cost dial — dropping debug at the source removes the largest volume bucket before serialisation or shipping, so the saving is real cost, not just hidden display. The floor is a deployment/run dial, not a code change.
• Per-run debug boost — a control table that raises the floor to debug for one correlation_id gives deep visibility into a single investigation without turning debug on globally and drowning the index.
• Allow-list redaction pre-ship — redacting by key with an allow-list, in the chain before the line leaves the process, makes "no PII in the index" a structural guarantee rather than a hopeful regex. Unknown fields drop by default.
• Correlation-coherent tail sampling — keying the sample on correlation_id and deciding at the tail keeps whole runs and 100% of failures, so sampling cuts volume without shredding the correlation that makes logs useful.
• Cost — O(1) per line for the level check and redaction, and O(lines per run) buffering for the tail decision. The eliminated cost is O(volume) index/storage spend plus the unbounded liability of PII in a searchable index. Net: constant per-line overhead plus bounded per-run buffering buys a large volume cut and a hard PII guarantee.

Logs
Topic — log-processing
Log-processing problems on levels and sampling

Practice →

Data Topic — etl ETL problems on volume and redaction

Practice →

---

5. Shipping and querying logs

stdout to collector to index to dashboard — structured fields become metrics, alerts, and SLOs for free

The mental model in one line: shipping is the path a structured log event walks from the process that emits it to a place where it can be queried — the application writes JSON to stdout, a collector (Fluent Bit, Vector, the platform agent) parses and batches it, a log aggregation platform indexes the fields, and a dashboard queries them — and because the events are structured, the same fields drive not just search but metrics, alerts, and SLOs, turning logs into a first-class observability signal rather than a debugging afterthought. Logs that are perfectly structured but never shipped, or shipped but never indexed on the right fields, are worthless; the last mile is where structured logging pays off or fails.

The shipping path — four hops, each with one job.

• Emit to stdout. The process writes one JSON object per line to stdout/stderr. It does not manage files, rotation, or network — that is the platform's job (Twelve-Factor logs). One line, one event, one newline.
• Collect. A node-level agent (Fluent Bit, Vector, Filebeat, the cloud log agent) tails the container's stdout, parses the JSON, adds infrastructure metadata (host, pod, container), batches, and forwards. This is where a broken JSON line is caught and where labels are attached.
• Index. The aggregation platform ingests the batches and indexes the low-cardinality fields as labels (service, event, level) while keeping the full JSON body searchable. Index design here decides query speed and cost.
• Query and visualise. A dashboard queries the index — filter by correlation_id, group by event, count by level — and renders panels, alerts, and SLO burn-down. This is the payoff the whole chain exists for.

Indexing and cardinality on the platform side.

• Labels vs body. Platforms like Loki index a small set of labels (streams) and keep the rest of the line as searchable content. Putting a high-cardinality field (correlation_id, order_id) in a label explodes the number of streams and destroys performance — keep those in the body and filter with a line search.
• The cardinality bomb. The single most common self-inflicted outage is promoting a high-cardinality field to an index label. Millions of label values create millions of tiny streams; ingestion and query both collapse. Keep labels to bounded dimensions.
• Retention tiers. Hot (searchable, expensive, days), warm (slower, weeks), cold/archive (object storage, cheap, months). Route by importance: error logs to a longer tier, sampled info to a short one.
• Parse at the edge, not at query time. Parsing JSON once at the collector and indexing fields is far cheaper than re-parsing free text on every query. This is the structural reason JSON logs beat regex-extracted plaintext at scale.

Metrics, alerts, and SLOs from structured fields.

• Metrics from logs. count by (event) where level="error" is an error-rate metric; quantile(fields.duration_ms, 0.95) by service is a latency SLI — both derived from the same structured fields, no separate instrumentation needed.
• Alerts on fields. Alert when count(level="critical") > 0, when error rate for a service exceeds a threshold, or when a run's duration_ms p95 breaches an SLO. The alert queries structured fields, so it is precise, not a substring match.
• The correlation-id drill-down. An alert fires with a correlation_id; one click filters the whole run's logs (and, if trace_id is present, the distributed trace). Structured logging is what makes alert-to-root-cause one hop.
• SLOs. Define the SLI as a query over structured fields (success rate = 1 - errors/total), set the objective, and burn down error budget from the same log stream. Logs, metrics, and SLOs share one source of truth.

Common interview probes on shipping and querying.

• "What ships the logs?" — a node collector (Fluent Bit/Vector) tailing stdout; the app doesn't ship.
• "What's the cardinality bomb?" — a high-cardinality field promoted to an index label; keep ids in the body.
• "How do you get a metric from logs?" — aggregate a structured field (count by event, quantile(duration_ms)).
• "How does an alert reach root cause fast?" — the alert carries the correlation_id; one filter shows the whole run.

Worked example — a Vector collector config that parses and routes

Detailed explanation. The collector is where raw stdout becomes indexed events. Build a Vector pipeline that parses the JSON, drops any line that failed redaction (defense in depth), routes error+ to a long-retention sink, and forwards the rest to the primary index — all keyed off the structured fields.

• Source. Tail container stdout.
• Transform. Parse JSON; add host/pod labels; guard against unparseable lines.
• Route. error+ → long-retention sink; everything else → primary.

Question. Write a Vector config that parses JSON logs, attaches infrastructure labels, and routes by level to two sinks.

Input.

Stage	Job
source	tail stdout / container logs
parse	JSON → structured event
enrich	add host, pod, container
route	error+ → archive; rest → primary

Code.

# vector.toml — parse, enrich, route structured logs by level
[sources.pipeline_stdout]
type = "file"
include = ["/var/log/pods/*/*/*.log"]

[transforms.parse_json]
type = "remap"
inputs = ["pipeline_stdout"]
source = '''
  . = parse_json!(.message)          # raw line -> structured fields; drops on bad JSON
  .host = get_hostname!()            # enrich with infra metadata
  .ingest_ts = now()                 # ingest time, SEPARATE from event ts
'''

[transforms.route_by_level]
type = "route"
inputs = ["parse_json"]
route.errors = '.level == "error" || .level == "critical"'
route.normal = '.level == "info"  || .level == "warn" || .level == "debug"'

[sinks.archive_errors]                # long retention for failures
type = "loki"
inputs = ["route_by_level.errors"]
endpoint = "http://loki:3100"
labels = { service = "{{ service }}", level = "{{ level }}" }   # LOW-cardinality labels only

[sinks.primary_index]
type = "loki"
inputs = ["route_by_level.normal"]
endpoint = "http://loki:3100"
labels = { service = "{{ service }}", level = "{{ level }}" }
# correlation_id / order_id stay in the BODY, never a label (cardinality bomb)

Step-by-step explanation.

1. The file source tails the container log files where the runtime writes each pod's stdout. The application itself ships nothing — it only wrote JSON to stdout, exactly as Twelve-Factor prescribes.
2. parse_json! turns each raw line into structured fields; the ! form drops lines that aren't valid JSON, so a malformed line can't poison the index. This is where the "one JSON object per line" contract is enforced at ingest.
3. Enrichment adds host and a separate ingest_ts. Keeping ingest_ts distinct from the event ts is essential: late-arriving logs must not overwrite when the event actually happened. Conflating the two is a classic debugging trap.
4. route_by_level splits the stream: error/critical to archive_errors (long retention, because failures are investigated for weeks) and everything else to primary_index (short retention). Routing is driven by the structured level field, not a substring guess.
5. Both sinks label only service and level — bounded, low-cardinality dimensions. correlation_id and per-record ids deliberately stay in the body, searchable but not indexed as labels, avoiding the cardinality bomb that would create millions of streams.

Output.

Line	Parsed?	Routed to	Labels
valid info JSON	yes	primary_index	service, level
valid error JSON	yes	archive_errors	service, level
malformed line	no (dropped)	—	—
correlation_id field	yes	in body	not a label

Rule of thumb. Parse JSON once at the collector, enrich with infra metadata, keep event time and ingest time separate, and route by the structured level. Label only bounded dimensions (service, level); keep every high-cardinality id in the searchable body to avoid the cardinality bomb.

Worked example — a LogQL query and a metric-from-logs panel

Detailed explanation. Once indexed, the payoff is queries that would be impossible over free text. Build three: filter one run by correlation_id, compute an error-rate metric per service, and compute a p95 duration SLI — all from the same structured fields, in LogQL (the pattern maps directly to OpenSearch/CloudWatch Insights).

• Drill-down. Filter service and line-match the correlation_id.
• Error-rate metric. count_over_time of error events by service.
• Latency SLI. quantile_over_time of fields.duration_ms.

Question. Write LogQL for a run drill-down, a per-service error-rate metric, and a p95 duration SLI over structured fields.

Input.

Query	Fields used
run drill-down	service (label), correlation_id (body)
error rate	service, level
p95 duration	fields.duration_ms

Code.

# 1. Drill-down: every line for one run. service is a label; correlation_id is in the body.
{service="orders-etl"} | json | correlation_id="manual__2026-09-05T02:14:00"

# 2. Error-rate metric: errors per minute per service (a metric derived from logs).
sum by (service) (
  count_over_time(
    {service=~".+"} | json | level="error" [1m]
  )
)

# 3. p95 task duration SLI: percentile over a structured numeric field.
quantile_over_time(0.95,
  {service="orders-etl"} | json | unwrap duration_ms [5m]
) by (service)

Step-by-step explanation.

1. Query 1 selects the orders-etl stream by its service label (cheap, indexed), then | json parses the line body and correlation_id="..." filters to one run. Because correlation_id lives in the body, it is a content filter, not a label — which is exactly right for a high-cardinality field.
2. Query 2 turns logs into a metric: count_over_time(... level="error" [1m]) counts error events per minute, and sum by (service) aggregates per pipeline. This error-rate metric needs no separate counter instrumentation — it falls out of the structured level field.
3. Query 3 computes a latency SLI: unwrap duration_ms lifts the numeric field out of the JSON body, and quantile_over_time(0.95, ...) computes p95 over a 5-minute window. The measure exists only because duration_ms was logged as a number, not a string — the type decision from section 2 pays off here.
4. All three queries share one substrate: the same structured events, indexed once. Search, metrics, and SLIs are three views of the same stream, so they never disagree the way separately-instrumented signals do.
5. The pattern is portable: OpenSearch's stats count() by service, CloudWatch Insights' stats pct(duration_ms, 95), and Datadog's log-based metrics express the same three queries. Structured fields are what make each platform's query language reach them.

Output.

Query	Result shape
drill-down	every line for manual__2026-09-05T02:14:00, time-ordered
error rate	errors/min per service, a graph line
p95 duration	p95 duration_ms per service, an SLI panel

Rule of thumb. Filter by indexed labels first (service), then refine on body fields (correlation_id, duration_ms) with | json. Derive metrics and SLIs from the same structured fields you search — one stream, three views, zero drift between them.

Worked example — an alert wired to a correlation-id drill-down

Detailed explanation. The final payoff is closing the loop: an alert fires on a structured-field condition and carries the correlation_id so the on-call engineer is one click from the whole run. Build an alert rule that triggers on error rate and templates the drill-down link into the notification.

• Condition. Error rate for a service exceeds threshold over a window.
• Payload. Include the offending service and a sample correlation_id.
• Link. Template a dashboard URL pre-filtered to that run.

Question. Define an alert rule that fires on per-service error rate and delivers a one-click drill-down to the failing run.

Input.

Element	Value
condition	errors/min by service > 10 for 5m
annotation	service, sample correlation_id
link	dashboard URL filtered to correlation_id

Code.

# alert-rules.yaml — fire on structured error rate; hand over the correlation id
groups:
  - name: pipeline-structured-logs
    rules:
      - alert: HighPipelineErrorRate
        expr: |
          sum by (service) (
            count_over_time({service=~".+"} | json | level="error" [5m])
          ) > 50
        for: 5m
        labels:
          severity: page
        annotations:
          summary: "High error rate in {{ $labels.service }}"
          # The drill-down: pre-filtered dashboard for the failing service.
          # On-call clicks through, then filters by the correlation_id in the panel.
          drilldown: "https://logs.internal/d/pipeline?var-service={{ $labels.service }}&level=error"
          runbook: "Filter the panel by correlation_id, read fields.reason, check the failing task."

Step-by-step explanation.

1. The expr is the same error-rate metric from the previous example, thresholded: if a service emits more than 50 error events over 5 minutes, the condition is met. It queries the structured level field, so it fires on genuine errors, not on the substring "error" appearing in some payload.
2. for: 5m requires the condition to hold for five minutes, filtering transient blips. severity: page routes it to the on-call pager. Both are driven by the structured metric, so the alert is precise.
3. The summary annotation templates the offending service label into the message, so the pager notification already says which pipeline is failing — no triage needed to identify the service.
4. The drilldown annotation templates a dashboard URL pre-filtered to the failing service and level=error. The on-call engineer clicks straight to the failing lines, then filters by a correlation_id shown in the panel to isolate one run and read fields.reason.
5. This is the loop structured logging exists to close: alert → service → run → root cause, each step a filter on a structured field. Without structure, the alert would be a substring count and the drill-down a grep; with it, root cause is a few clicks from the page.

Output.

Step	Engineer sees
page fires	"High error rate in orders-etl"
click drilldown	failing error lines for orders-etl
filter correlation_id	one run's whole narrative
read fields.reason	root cause, no grep

Rule of thumb. Alert on structured-field conditions, template the failing service into the notification, and hand over a pre-filtered drill-down link. The measure of a good logging setup is the number of clicks from page to root cause — structured fields make it two or three, free text makes it an afternoon.

Data engineering interview question on end-to-end shipping and querying

A senior interviewer might ask: "Design end-to-end log shipping and querying for a platform of 100 pipeline tasks emitting JSON to stdout. Cover the collector, the indexing strategy and how you avoid a cardinality blow-up, retention tiers, how you derive an error-rate metric and a latency SLO from the logs, and how an alert gets an on-call engineer from page to root cause in under a minute."

Solution Using stdout → Vector → aggregation platform with label discipline and log-derived SLOs

# 1. Vector: parse, enrich, route by level, LABEL ONLY low-cardinality fields.
[sources.stdout]
type = "kubernetes_logs"

[transforms.structure]
type = "remap"
inputs = ["stdout"]
source = '''
  . = parse_json!(.message)
  .ingest_ts = now()               # separate from event ts
'''

[transforms.route]
type = "route"
inputs = ["structure"]
route.errors = '.level == "error" || .level == "critical"'
route.rest   = 'true'

[sinks.hot_errors]                 # long retention tier for failures
type = "loki"
inputs = ["route.errors"]
endpoint = "http://loki:3100"
labels = { service = "{{ service }}", level = "{{ level }}", event = "{{ event }}" }

[sinks.primary]                    # short retention tier for the rest
type = "loki"
inputs = ["route.rest"]
endpoint = "http://loki:3100"
labels = { service = "{{ service }}", level = "{{ level }}" }
# correlation_id, order_id, trace_id stay in the body — NEVER labels.

# 2. Log-derived error-rate metric and latency SLO over structured fields.
# Error rate per service (metric from logs):
sum by (service) (count_over_time({service=~".+"} | json | level="error" [5m]))

# Latency SLO SLI — p95 task duration per service:
quantile_over_time(0.95, {service=~".+"} | json | unwrap duration_ms [5m]) by (service)

# Success-rate SLI for the error budget (1 - errors/total):
1 - (
  sum(count_over_time({service="orders-etl"} | json | level="error" [30d]))
  /
  sum(count_over_time({service="orders-etl"} | json | event="run_finished" [30d]))
)

# 3. Alert + drill-down: page to root cause in one hop.
- alert: PipelineErrorSpike
  expr: sum by (service) (count_over_time({service=~".+"} | json | level="error" [5m])) > 50
  for: 5m
  annotations:
    summary: "{{ $labels.service }} error spike"
    drilldown: "https://logs.internal/d/pipe?var-service={{ $labels.service }}&level=error"

Step-by-step trace.

Layer	Decision	Reasoning
Emit	JSON to stdout	Twelve-Factor; app ships nothing
Collect	Vector kubernetes_logs	parse once, enrich, route
Labels	service, level, (event for errors)	bounded cardinality; fast queries
Body	correlation_id, ids, duration_ms	high-cardinality stays searchable, not a label
Retention	hot errors (long), primary (short)	failures kept for weeks, info days
Metrics/SLO	log-derived error rate + p95 + success rate	one stream, no separate instrumentation
Alert	structured level threshold + drilldown	page → service → run → reason

After deployment, 100 tasks emit JSON to stdout; Vector parses each line once, routes errors to a long-retention tier and the rest to a short one, and labels only bounded dimensions so the platform never suffers a cardinality blow-up. Error-rate metrics, a p95 latency SLO, and a 30-day success-rate SLI all derive from the same structured stream. An alert fires on the structured level field, names the failing service, and links a pre-filtered dashboard, so on-call goes from page to fields.reason in under a minute.

Output:

Metric	Value
Ingest parse	once, at the collector
Index labels	2–3 low-cardinality fields
Cardinality-bomb risk	none (ids in body)
Retention	errors long, info short
Signals from logs	search + metrics + SLOs
Page → root cause	< 1 min (drill-down link)

Why this works — concept by concept:

• stdout + collector split — the app emits JSON and nothing else; the collector parses, enriches, and ships. This Twelve-Factor split means the pipeline code never deals with files, rotation, or network, and the collector centralises parsing and routing.
• Label discipline — indexing only service, level, and (for errors) event keeps stream cardinality bounded, so ingestion and queries stay fast. Keeping correlation_id and per-record ids in the searchable body avoids the cardinality bomb that fells naive setups.
• Retention tiers by severity — routing error+ to a long tier and the rest to a short one matches spend to investigative value: failures are read for weeks, routine info for days.
• Signals from one stream — error-rate metrics, a p95 SLO, and a success-rate SLI all derive from the same structured fields, so search, metrics, and SLOs never disagree — one source of truth instead of three instrumentations.
• Cost — O(1) parse and index-write per line over bounded labels, and O(matching lines) queries over an index rather than O(all lines) over free text. The eliminated cost is the cardinality-bomb outage and the separate metric/SLO instrumentation. Net: disciplined labels and one structured stream buy search, metrics, alerts, and SLOs at the price of one ingest parse per line.

Logs
Topic — log-processing
Log-processing problems on aggregation and queries

Practice →

Data
Topic — etl
ETL problems on log-derived metrics

Practice →

---

Cheat sheet — structured logging recipes

• Event, not string. Every log line is a JSON object; name the low-cardinality event once (rows_processed, order_validation_failed) and lift every variable into a typed field. If you ever need a regex against your own logs, you logged them wrong. sum(rows) by table should be a query, never an awk script.
• Reserved-key schema. Fix a small vocabulary on every line: ts (event time, one clock, RFC3339/epoch), level (enum debug|info|warn|error|critical), event (stable name), service, run_id/correlation_id, optional trace_id, and a nested fields bag for everything event-specific. Reserved keys are sacred — never let an event field collide with them.
• Types and cardinality. Numbers stay numbers so aggregates work; the classic bug is stringifying everything. Index only low-cardinality dimensions (service, event, level) as labels; keep high-cardinality ids (correlation_id, order_id) in the searchable body — promoting one to a label is the cardinality bomb.
• One event, two renderers. Compact single-line JSON to stdout in prod; pretty HH:MM:SS level event key=value for humans in dev. Same event dict, one call site, renderer chosen by env var. Never maintain two logging code paths.
• structlog processor chain. merge_contextvars → add_log_level → TimeStamper(fmt="iso", key="ts") → EventRenamer("event") → JSONRenderer (prod) / ConsoleRenderer (dev). Enforce the schema in the chain, not at each call site, so there is zero drift.
• Additive-only schema evolution. Add keys and enum values; never rename or retype. Dual-write old + new for a deprecation window at least as long as log retention, migrate consumers, then drop. Guard reserved keys with a CI schema check so a breaking change fails the build, not the 3 a.m. dashboard.
• Correlation-id contextvar pattern. Generate-or-adopt one id at the true edge, bind it to a contextvars.ContextVar inside a scope with a guaranteed reset, and let the formatter stamp it on every line. Never pass the id as a function argument; never store it in a plain global (concurrent runs would bleed).
• Propagate across every boundary. In-process: contextvars (inherited across await/threads). Orchestrator: reuse the run_id across all tasks and retries. Subprocess: env var / CLI arg. Kafka/SQS/Pub-Sub: message header, bound on receive. HTTP/gRPC: W3C traceparent header. Id in the header, contextvar in the process.
• id taxonomy. correlation_id = business/run scope (yours). trace_id = W3C distributed-trace root (32 hex). span_id = one operation (16 hex). traceparent = version-traceid-spanid-flags wire format. Carry correlation_id + trace_id on the same lines so logs and traces cross-link.
• Levels route and cost. debug off in prod (dynamic per-run boost via a control table), info the narrative, warn a leading indicator (never sampled), error/critical always kept and duplicated to the alert sink. Route by level in one place, not with if at every call site.
• Correlation-coherent tail sampling. Key the sample on hash(correlation_id) % 100 < pct so whole runs are kept or dropped; decide at the tail so you keep 100% of failed runs and 1% of successful ones. Never sample randomly per line (shreds runs) and never sample away a warn+.
• Allow-list PII redaction. Declare allowed field names; mask (a***@domain), hash (sha256 for joinable-but-not-reversible), or drop everything else, in a processor that runs before the line ships. Allow-list beats deny-list — it stays safe when someone adds a new PII field upstream without telling you.
• Ship stdout → collector → index → dashboard. App writes JSON to stdout (Twelve-Factor); a collector (Vector/Fluent Bit) parses once, enriches with infra metadata, keeps event ts separate from ingest_ts, and routes by level to retention tiers. Metrics (count by event where level="error"), latency SLIs (quantile(duration_ms, 0.95)), and success-rate SLOs all derive from the same structured stream; alerts template the failing service and a pre-filtered correlation_id drill-down so on-call goes page → root cause in one hop.

Frequently asked questions

What is structured logging in one sentence?

Structured logging is the practice of emitting every log line as a machine-parseable event — a set of typed key-value fields, almost always a single JSON object with a stable schema — instead of a human-readable sentence, so that downstream tooling can filter, group, aggregate, and alert on individual fields rather than running substring searches over free text. In a data pipeline that means one JSON object per event with reserved keys (ts, level, event, run_id, and a nested fields bag), so questions like "which run failed, on which record, with what error" become one query instead of an afternoon of grep and awk. It is the contract that makes log aggregation, dashboards, metrics-from-logs, and alerting all work.

Structured logging vs plain text — when do I switch?

Switch the moment your pipeline has more than one task, more than one worker, or more than one concurrent run — because the questions you need to answer during an incident ("which run? which task? which record? how many?") are field questions, and plain text has no fields. Plain text is fine for a single script a human watches in a terminal; it collapses the instant logs from many sources interleave. The cheapest first step is not even full JSON: add a correlation_id/run_id to every line so you can at least grep one run out of the interleaved stream, then convert to JSON events (with PII redaction) so the same questions become queries rather than searches. Keep a pretty renderer for local development so humans still get readable lines — the event is structured either way; only the surface form changes.

What is a correlation id vs a trace id?

A correlation_id is a business/run-level identifier you generate to answer "show me everything about this one unit of work" — one per pipeline run, request, or message, propagated unchanged through every task, retry, and hop. A trace_id is the distributed-tracing root identifier defined by W3C Trace Context (16 bytes, 32 hex characters), shared by every span of one distributed trace and understood natively by tracing tools like OpenTelemetry, Jaeger, and Tempo. They serve overlapping purposes at different scopes: the correlation id is yours (you name it, often reusing the orchestrator's run_id), while the trace id is a standard that crosses service boundaries via the traceparent header. The mature setup logs both on the same lines so a log filter and a trace view cross-link, and a span_id pins a line to one exact operation.

How do I propagate a correlation id across pipeline tasks?

Generate (or adopt an inbound) id exactly once at the true edge, bind it to a contextvars.ContextVar so every log call in that context inherits it without argument threading, and reset the binding when the unit of work ends so concurrent runs never bleed. Across an orchestrator like Airflow, reuse the DAG run_id as the correlation id and bind it at the start of every task — because retries keep the same run_id, all attempts stay joined. Across a queue (Kafka, SQS, Pub/Sub) put the id in a message header (not the body) and bind it on the consumer side before processing. Across HTTP/gRPC, carry the W3C traceparent header and extract it server-side. The rule everywhere is the same: id travels in the header on the wire and in a contextvar inside the process, so one filter on correlation_id returns every line the run produced.

How do I keep PII out of logs?

Redact by field key with an allow-list, in a processor that runs before the line ever leaves the process. Declare which field names are allowed in logs (safe measures like rows, table, duration_ms); for everything else, mask for partial visibility (a***@example.com), hash when you still need to join or count distinct without the raw value (sha256(email)), or drop the field entirely. An allow-list beats a deny-list because it stays safe when someone adds a new PII field upstream — unknown keys drop by default rather than leaking until you notice. Because the data is structured, redaction targets fields deterministically by name instead of scrubbing free text with regexes that always miss one format. Crucially, redact at the earliest point in the logging chain: PII must never reach the collector, index, or disk, because once persisted to a searchable index it is a compliance incident, and redacting downstream is too late.

Structured logs vs metrics vs traces — which do I need?

You need all three, and structured logging is what makes them reinforce each other rather than drift apart. Metrics are cheap aggregate time series (error rate, throughput, p95 latency) — great for dashboards and alerts but with no per-event detail. Traces show the causal path of one request across services with timing per span — great for latency root-cause but sampled and not a full record. Structured logs are the per-event record with full context — great for "exactly what happened to this record in this run." The connective tissue is shared identifiers: a trace_id and correlation_id on your structured log lines let an alert (metric) drill into a trace and into the exact log lines for one run, and you can even derive metrics and SLIs from structured log fields (count by event where level="error", quantile(duration_ms, 0.95)) so one stream feeds search, metrics, and SLOs without three separate instrumentations.

Practice on PipeCode

• Drill the log-processing practice library → for the event-parsing, correlation-join, level-filtering, and aggregation problems senior interviewers use to probe observability skills.
• Rehearse on the etl practice library → for the multi-task pipeline, sampling, and log-derived-metric patterns that turn structured events into dashboards.
• Sharpen your parsing intuition on the JSON practice library → for the schema-design, type-discipline, and nested-field problems that make JSON logs query-able.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the reserved-key schema, correlation-id propagation, and shipping pipeline against real graded inputs.

Lock in structured logging muscle memory

Docs explain the fields. PipeCode drills explain the decision — when a plaintext line should have been an event, where the correlation id has to be generated, why a high-cardinality field must stay out of the index labels, how a level floor and tail sampling cut cost without losing a single failure. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face.

Practice log-processing problems →
Practice etl problems →