Ingestion & Data Pipelines

Three patterns, nine sources

Chapter 01 ended on a single load-bearing sentence: three ingest patterns cover all nine GridDP sources — log-based CDC for mutable OLTP, streaming for append-only events, and batch API pull for third-party systems. The firehose is "streaming, routed to a specialist store." This chapter fills in the mechanics, but the organizing principle never changes:

Here is the master map from chapter 01, collapsed to the one column this chapter is about — and the section that designs each:

The rest of the chapter walks these three patterns in depth, then the cross-cutting concerns every pipeline shares: ELT-vs-ETL placement, schema evolution, and reliability.

Log-based Change Data Capture

The four OLTP sources (marketplace-api, billing-service, instance-orchestrator, ranking-service) are mutable: an offer.price changes, a rental.status walks pending → active → stopped. The naïve way to ingest them is a nightly SELECT * snapshot. The senior way is log-based CDC. The difference is not cosmetic — it is the difference between a platform that can answer "what was the price at 14:00?" and one that can only ever see the latest state.

Why CDC beats nightly SELECT *

• History. A nightly snapshot sees only the row's final state for the day. If an offer's price changed five times between snapshots, four of those states are gone forever. CDC streams every insert/update/delete, so the platform reconstructs the full timeline — the raw material for slowly-changing dimensions (chapter 06).
• No OLTP load. A full-table SELECT * scans every row and competes with production traffic for buffer cache and I/O. Log-based CDC reads the database's write-ahead log — a sequential file the database already writes for crash recovery — so it imposes near-zero load on query paths.
• Deletes are visible. A snapshot diff can infer a delete (a row that was there yesterday is gone today) but cannot tell a delete from a filtered query or a transient outage. The WAL records the delete explicitly.

How Debezium reads the Postgres WAL

Postgres exposes its WAL through logical replication: you create a replication slot and a publication, and Postgres decodes committed changes into a logical stream of row-level events. Debezium connects as a replication client, consumes that stream, and publishes one Kafka message per change — typically one topic per table.

A Debezium change event carries a before image, an after image, the operation (create / update / delete / read-snapshot), and source metadata including the log position (Postgres LSN) and commit timestamp. That LSN is the offset that makes the stream replayable and exactly-once-friendly.

{
  "name": "marketplace-api-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "marketplace-db.internal",
    "database.dbname": "marketplace",
    "plugin.name": "pgoutput",
    "slot.name": "griddp_cdc",
    "publication.name": "griddp_pub",
    "table.include.list": "public.offer,public.rental,public.bid,public.account",
    "snapshot.mode": "initial",          // snapshot once, then tail the WAL
    "tombstones.on.delete": "true",       // emit a null-value tombstone after a delete
    "topic.prefix": "griddp",
    "decimal.handling.mode": "precise",   // money must not become a float
    "heartbeat.interval.ms": "10000"      // advance the slot even on idle tables
  }
}

Initial snapshot, then tail the stream

A connector cannot start from "now" — the lake would be missing every row that existed before CDC turned on. So the lifecycle is two phases. First, an initial snapshot: Debezium reads the current table contents as a batch of synthetic read events, establishing the baseline. Then it switches to streaming, tailing the WAL from the LSN captured at snapshot time. Done right the two phases are seamless — no gap, no overlap — because the snapshot LSN is the exact handoff point. The heartbeat.interval.ms above matters here: on a low-traffic table the slot would otherwise never advance, and Postgres would retain WAL indefinitely until the disk fills.

Deletes and tombstones

A delete in Postgres becomes a Debezium event with a populated before and a null after. With tombstones.on.delete enabled, Debezium then emits a second message with a null value on the same key — a tombstone. Kafka log compaction uses the tombstone to physically drop the key. Downstream, the platform must treat the delete as data, not absence: a deleted offer is not "never existed," it is "withdrawn at time T," which the reliability and liquidity metrics genuinely care about.

From CDC events to SCD2 history

The reason CDC is worth the operational weight is what it enables downstream. Each change event is timestamped with its commit time; a stream of them for one key is a history. In chapter 06 we turn that history into a slowly-changing dimension type 2 — one row per version of an entity, with valid_from / valid_to bounds — so an analyst can ask "what was this offer's price when that rental started?" and get the right answer. CDC is the ingestion-layer prerequisite for that; without it the question is unanswerable.

Event streaming & watermarks

The append-only sources — marketplace-events, the trust-safety signal stream, and orchestrator lifecycle events — are published to Kafka. Streaming ingestion looks simple ("just read the topic") and is full of correctness traps. Two concepts decide whether you get it right: delivery semantics and event-time handling.

Topics, partitions, keys

A Kafka topic is an append-only log split into partitions for parallelism. Ordering is guaranteed only within a partition, and a message's partition is chosen by its key. For GridDP the key choice is a modeling decision: key bid.placed by account_id and all of one account's bids stay ordered and land on one consumer; key by offer_id and you can reconstruct a single offer's bid history in order. The data contract from chapter 01 pinned this down — ordering: per-account_id.

{
  "key": "acct_91f2",
  "headers": { "schema_id": 412, "content_type": "avro/binary" },
  "value": {
    "event_id":    "5b1c…-uuid",          // idempotency / dedupe key
    "event_type":  "bid.placed",
    "event_time":  "2026-06-16T14:03:07.220Z",  // producer-stamped, UTC
    "account_id":  "acct_91f2",
    "offer_id":    "offer_77c0",
    "bid_usd_hr":  0.41,
    "is_interruptible": true
  },
  "partition": 3,
  "offset": 88412290
}

Delivery semantics & idempotency

Kafka, like most streaming systems, defaults to at-least-once delivery: on retry or rebalance a message can be redelivered, so a consumer that naïvely inserts will double-count. Exactly-once is achievable (Kafka transactions, idempotent producers, transactional sinks) but it is expensive and end-to-end only as strong as its weakest hop. The pragmatic GridDP default: at-least-once delivery plus idempotent consumers. Because every event carries an event_id (mandated by the contract), the sink dedupes on it — a redelivered bid.placed is dropped, and effective semantics are exactly-once without the full transactional machinery.

Event-time vs processing-time — the crucial distinction

Chapter 01 flagged this with the bid.placed example: a bid can arrive at the platform seconds or minutes after it occurred — a mobile client was offline, a retry fired, the broker lagged. There are two clocks:

• Event time — when the thing actually happened, stamped by the producer (event_time in the message above).
• Processing time — when the platform happened to read it.

Bucket your "bids per minute" chart by processing time and a flurry of late-arriving bids all pile into the wrong minute — the chart is wrong, and the interruptible-clearing-price model trains on corrupted features. Correct windowing always keys on event time. But that raises the question: if data can arrive late, when is a time window "done" and safe to emit?

Watermarks & late data

A watermark is the stream processor's assertion: "I believe I have now seen all events with event_time ≤ T." It trails real time by a configured allowed lateness (say, 2 minutes). When the watermark passes a window's end, the window fires and emits its aggregate. Events that arrive after the watermark are late and handled by an explicit policy — drop, route to a side output, or trigger a window update — never silently mis-bucketed.

The allowed-lateness knob is a direct latency-vs-correctness trade. A long horizon catches the flaky-residential-internet stragglers (picture residential, hobbyist-grade providers) but delays every result; a short horizon is snappy but drops more late data. There is no universally right value — it is a per-pipeline decision driven by how much lateness the source actually exhibits and how much the consumer can tolerate.

Designing the telemetry firehose

Now the source that dominates the platform: ~17,000 GPUs × one sample / 10 s = ~1,700 samples/sec = ~150 M/day. It is technically "streaming," but its volume and its dual use-case force a design the other event streams do not need.

The dual-path pattern

The telemetry stream serves three SLAs at once (chapter 01): operational dashboards that ask "is the fleet healthy now?", ML features that predict reliability and detect throttling, and a billing input that proves a GPU was delivering during a rented hour. No single store serves all three well. So the stream forks: one path to a time-series engine for real-time ops, one path to object storage for analytics and ML.

The two paths have opposite optimizations. The ops path into ClickHouse optimizes for freshness and cheap recent-window queries — it is the "is the fleet on fire right now" path, and it intentionally retains only a short window of fine-grained data. The analytics path optimizes for cost-per-byte-at-rest and replayability: it lands the raw samples untouched so any future ML feature can be recomputed, then derives the small per-GPU-hour rollups that flow into the warehouse for joining with business data.

Batching, compression, partitioning

Writing 1,700 tiny rows per second straight to object storage is a disaster — millions of small files, ruinous request costs, unusable scan performance (the "small files problem," chapter 05). The analytics path therefore buffers samples and writes them in batches — e.g. flush every 60 seconds or every 128 MB, whichever comes first — as columnar Parquet with compression. Files are partitioned by time (dt=…/hr=…) and bucketed by machine_id, because nearly every query is "metric for entities over a time window" and time-partitioning lets the engine prune to the relevant hours.

Backpressure & buffering

The firehose is also where backpressure bites. If a sink slows (ClickHouse compaction, an object-store throttle), the system must not push back into the host-agents — they cannot buffer indefinitely on a hobbyist's home box, and lost telemetry corrupts billing and reliability. Kafka is the shock absorber: it durably retains the stream, so a temporarily slow consumer simply falls behind and catches up by replaying from its committed offset. Consumer lag (offset distance behind the head) is therefore the firehose's vital sign — the metric the on-call watches, covered under reliability below.

Third-party batch ingestion

The third-party sources — payments (Stripe, crypto rails, KYC/AML) and support (Zendesk-like) — are pulled, not pushed. The company does not own them, so they come with rate limits, pagination, and schemas that change on the vendor's schedule. Two patterns cover them.

Webhook for timeliness, batch pull for truth

Stripe (chapter 01) illustrates the canonical pairing. A webhook fires the moment a charge succeeds — timely, but webhooks can be missed (your endpoint was down, the delivery was dropped). A nightly API reconciliation pull re-fetches the day's objects and is the source of truth. The webhook gives you low latency; the batch pull guarantees completeness. You need both, and they must converge — the same charge arriving via both paths is deduped on its Stripe object id (idempotency again).

Pagination, rate limits, incremental extraction

A batch puller is mostly plumbing, and the plumbing is where it breaks:

• Pagination. APIs return pages via cursors or offsets; the puller must follow them to completion and survive resuming mid-walk.
• Rate limits. Vendors throttle. The puller respects 429 / Retry-After with backoff, or it gets blocked and the pull fails half-done.
• Incremental high-watermark extraction. Never re-pull all history every night. Track the maximum updated_at (or a cursor) from the last successful run and request only records changed since — the high-watermark. This bounds cost and runtime as the vendor's data grows. (Note: this is the extraction high-watermark, a different mechanism from the streaming watermark in the firehose section, which governs window completeness.)

Connector tools

Hand-writing pullers for dozens of SaaS APIs does not scale, so managed connectors exist precisely for this pattern. In Stack C, Fivetran handles pagination, rate limits, incremental extraction, and schema drift for support and payments out of the box; in Stack B the equivalent is Fivetran or Airbyte feeding Auto Loader. Stack A leans on Airbyte or bespoke extracts — cheaper in license, dearer in engineering time. The trade-off (buy vs. build the connector) is the same capital-vs-headcount choice chapter 03 framed for the whole stack.

ELT vs ETL — land raw, then transform

Every pattern above ends with data landing somewhere. What happens next — transform-then-load or load-then-transform — is the ELT/ETL question, and the modern default has flipped.

Modern stacks default to ELT: land raw to the lake or warehouse first, transform with SQL after. The reasons are decisive. Storage is cheap, so keeping the raw zone is affordable; warehouses are now fast enough to transform at scale; and — most importantly — retained raw means a logic change is a re-run, not a re-ingest. Discover a bug in how you derive a feature six months in, and with ELT you just re-run the transform over data you still have. With pure ETL, the untransformed truth may be gone.

ETL still earns its place at ingestion time for things that must happen before landing: redacting or tokenizing PII before it touches the lake (KYC/AML data, chapter 11), and heavy stream computation like the firehose's windowed rollups, which are cheaper to compute once in flight than repeatedly in SQL.

Schema registry & evolution

Chapter 01's anti-pattern list warned that a producing team can rename a column or add an enum value the moment a feature ships, and your pipeline breaks at 3 a.m. The data contract is the agreement; the schema registry is the runtime that enforces it.

Instead of embedding JSON shapes by convention, producers serialize events with Avro or Protobuf and register each schema in a central registry, which assigns a schema_id (note it in the Kafka message header above). Consumers fetch the schema by id to deserialize. Crucially, the registry rejects a producer's attempt to register a schema that violates the configured compatibility rule — so an incompatible change fails in the producer's CI, not in your pipeline at 3 a.m.

Handling drift without breaking the pipeline comes down to this: under a backward-compatible regime, a producer may add a new optional field and old consumers simply ignore it; a consumer may be upgraded to read the new field with a default for historical events that lack it. The pipeline keeps flowing throughout. A truly breaking change (renaming a required field, changing a type) is forced through the contract's breaking_change_policy — the major-version-bump + 2-week notice from chapter 01 — giving the platform time to migrate consumers. The registry is how a contract stops being a wiki page and becomes an enforced invariant; we wire it into CI and governance in chapters 09 and 10.

Pipeline reliability primitives

Ingestion runs unattended, so the design must assume every hop can fail, duplicate, or stall. A small set of primitives, applied consistently, is what separates a pipeline that pages weekly from one that self-heals.

• Idempotency. Every record carries a stable key (event_id, Stripe object id, CDC LSN). Sinks upsert/dedupe on it, so a replay or redelivery is harmless. This is the foundation everything else rests on.
• Exactly-once sinks. Where a duplicate is intolerable even transiently (the billing meter), use a transactional sink that atomically commits the write and the consumer offset together.
• Dead-letter queues (DLQ). A poison message — unparseable, fails the schema, violates a constraint — is routed to a DLQ instead of crashing the consumer or, worse, being dropped. The pipeline keeps moving; the bad records wait for inspection and replay. (The late-data side output from the streaming section is the watermark-world cousin of a DLQ.)
• Replay / reprocessing. Because Kafka retains the log and the lake retains raw (ELT), you can re-run a pipeline from a past offset or re-derive a table from raw after a logic fix — without re-ingesting from the source.
• Monitoring lag. The single best health signal for a streaming pipeline is consumer lag: how far behind the head the consumer is. Rising lag means a sink is struggling before any data is actually lost — the early warning. Pair it with freshness SLOs derived from the contract's freshness_p99.

How A/B/C each implement ingestion

The three patterns are universal; the tools differ by stack (chapter 03). Each stack expresses CDC, streaming/firehose, and batch differently:

Read across the rows and the chapter's thesis holds: every stack lands raw and transforms in-warehouse (ELT), every stack forks the firehose to a time-series engine, and every stack reaches for a managed connector for the messy third-party pulls. The differences are ergonomics and cost, not architecture — exactly the framing chapter 03 set up. The short version of the stack choice for ingestion: A assembles best-of-breed open-source (most control, most glue code), B unifies streaming and batch under Spark/DLT (strongest when ML and large-scale stream compute dominate), C minimizes ingestion engineering by leaning on Fivetran and Snowpipe (fastest time-to-value).

Takeaway

• Three patterns cover all nine sources: log-based CDC for mutable OLTP, streaming for append-only events, batch pull for third-party APIs — match the pattern to the source's shape, never the tool.
• CDC off the Postgres WAL (Debezium) preserves history and spares the OLTP database; the initial-snapshot-then-tail lifecycle and tombstones-as-data feed the SCD2 history of chapter 06.
• Streaming correctness lives in two ideas: idempotency (dedupe on event_id to make at-least-once effectively exactly-once) and event-time windowing with watermarks to handle late, out-of-order data without silent corruption.
• The firehose forks: stream to ClickHouse for real-time ops, land raw (batched, compressed, time-partitioned) to object storage for analytics/ML, and let only per-GPU-hour rollups into the warehouse.
• Default to ELT — land raw, transform in-warehouse — so a logic change is a re-run, not a re-ingest. A schema registry turns data contracts into enforced invariants, and idempotency + DLQs + replay + lag monitoring keep pipelines correct unattended.

We have the data flowing in. Next: where it lands and how it is laid out for cost and query speed. → Storage & the Lakehouse