11.28 Enterprise Data Ingestion & Analytics

Modern enterprises generate data at a staggering rate. Transactional databases (OLTP) are optimized for row-level writes and reads, but they collapse when asked to run complex analytical queries (OLAP) across millions of records. To solve this, we extract data from operational systems and move it into specialized analytical storage.

Requirements

The company here is a retailer with 40 PostgreSQL databases behind its services: orders, payments, inventory, customers, shipping and so on. Analysts and data scientists want all of it in one place, close to live.

Functional. Copy every insert, update and delete from the source databases into the analytics platform. Keep the raw history so anything can be reprocessed. Produce cleaned tables and business aggregates. Survive schema changes in the sources. Run backfills when logic changes. Check data quality. Keep personal data out of places it should not be.

Non-functional, with numbers.

• Freshness: a change in a source is in the cleaned (silver) tables within 5 minutes, at p95.
• The pipeline adds no more than 5% CPU load to any source database.
• No change is ever lost, and none is applied out of order.
• Any day can be reprocessed: 7 days from the log, any time from the raw (bronze) layer.

Out of scope today: the BI dashboards, and machine-learning feature pipelines built on top.

The clarifying questions, and what each answer changes

"How fresh is fresh?" "By tomorrow morning" is a nightly batch export and none of the streaming machinery below. "Within minutes" is CDC. The answer can delete most of this page, so ask it first.

"Do deletes matter?" In analytics they usually do (a cancelled order must leave the revenue figures). If yes, polling by updated_at is ruled out, because a deleted row leaves nothing to poll.

"Who owns the source schemas, and do they tell anyone before changing them?" If nobody coordinates, schema changes will break consumers weekly. The answer decides how strict the data contracts below must be.

"Which columns hold personal data?" Card numbers, addresses and phone numbers should be masked or dropped before the raw layer, because the raw layer is kept for years and copied often.

"What is the busiest day?" For a retailer it is a sale day, often several times a normal day. Capacity is planned for that day, not the average.

Estimation

Change rate. Across 40 databases, about 5,000 row changes a second on an ordinary day and 20,000 a second on the biggest sale day.

Bandwidth. Encoded as Avro with a registered schema, a change event is about 600 bytes. That is 5,000 × 600 = 3 MB a second on average and 12 MB a second at peak. What that forces: trivial for Kafka. The number of partitions is chosen for consumer parallelism and per-key ordering, not for bandwidth: around 12 partitions for the busiest topic lets 12 consumers work in parallel.

Daily volume. 5,000 × 86,400 seconds = 432 million events a day, about 260 GB of Avro. Stored in the bronze layer as Parquet, roughly 3 times smaller, it is about 86 GB a day, about 31 TB a year. What that forces: bronze lives in object storage partitioned by date, never in the warehouse, where storing every raw change would cost far more.

Log retention. Keeping 7 days in Kafka is 260 GB × 7 ≈ 1.8 TB, and with 3 replicas about 5.4 TB of disk. What that forces: the retention period is a cost decision. Thirty days would need about 23 TB; seven days plus the bronze layer gives the same ability to replay for less.

The number that causes real outages: write-ahead log held by a stuck connector. A CDC connector reads from a replication slot, and PostgreSQL keeps every byte of write-ahead log the slot has not yet confirmed. Assume about 2 KB of WAL per changed row (row data, index entries and page images; measure yours). At 5,000 changes a second that is about 10 MB a second, about 860 GB a day. What that forces: if the connector stops on Friday evening, the busiest source database can fill its disk over the weekend and stop accepting writes, an analytics problem turned into a production outage. Alert on replication-slot lag in bytes, and cap it with PostgreSQL's max_slot_wal_keep_size (PostgreSQL 13 and later), accepting that a capped slot must be re-snapshotted.

API

Most of this system talks through the log, so its most important "API" is the shape of a change event:

{ "schemaId": 412,
  "source": { "db": "orders-db", "table": "orders", "lsn": 98213441, "txId": 771201 },
  "op": "u",
  "ts_ms": 1760781667312,
  "before": { "id": "o_9182", "status": "PENDING", "total_cents": 459900 },
  "after":  { "id": "o_9182", "status": "SHIPPED", "total_cents": 459900 } }

The lsn is the position of this change in the source database's write-ahead log. It only ever increases within one database, which makes it the ordering key in the data model below.

A small control-plane API manages the pipeline:

POST /sources
{ "name": "orders-db", "tables": ["orders", "order_items"], "maskColumns": ["customers.phone"] }
→ 201 Created  { "sourceId": "src_14", "snapshot": "running" }

POST /contracts
{ "topic": "orders-db.orders", "schema": { …Avro… }, "compatibility": "BACKWARD" }
→ 201 Created
→ 409 Conflict  { "error": { "code": "INCOMPATIBLE_SCHEMA",
                             "message": "Field total_cents removed; consumers dim_orders, fraud_features depend on it.",
                             "requestId": "req_2a" } }

POST /backfills  { "target": "sales_gold", "from": "2026-01-01", "to": "2026-09-30", "logicVersion": "v2" }
→ 202 Accepted   { "backfillId": "bf_7", "writesTo": "sales_gold_v2" }
GET  /backfills/bf_7  → progress, row counts, validation results

A schema change is checked when it is registered, not when it breaks a consumer. The 409 names the consumers that depend on the removed field, which turns an incident next week into a conversation today.

A backfill never writes to the live table. The response says where it writes (sales_gold_v2), and the switch to the new table is a separate, deliberate step.

Data model

bronze.cdc_events                          -- raw, append-only, object storage
  ingest_date    DATE                      -- partition
  source_db, source_table TEXT
  lsn            BIGINT
  op             CHAR(1)
  before, after  JSON
  schema_id      INT

silver.orders                              -- one row per order, current state
  id             TEXT PRIMARY KEY
  status, total_cents, …
  _source_lsn    BIGINT NOT NULL           -- the LSN of the change this row reflects
  _deleted       BOOLEAN NOT NULL
  _updated_at    TIMESTAMPTZ

gold.sales_daily
  day DATE, region TEXT, revenue_cents BIGINT, orders INT
  PRIMARY KEY (day, region)

Access patterns:

Query	Frequency	Returns
Apply a batch of changes to silver	every minute per table	thousands of rows
Replay one table for a date range	backfills	millions of rows
Analysts' ad-hoc queries on silver and gold	constant	aggregates
Daily rebuild of gold	nightly	hundreds of rows

The _source_lsn column is what makes the merge correct, not just repeatable. The page says below that a MERGE on the primary key is idempotent, and it is: applying the same change twice gives the same row. But a replay or a snapshot can also deliver an older change after a newer one. A plain MERGE then overwrites "SHIPPED" with "PENDING", silently. Guarding the update with "only if the incoming lsn is greater than _source_lsn" makes old changes harmless, whatever order they arrive in:

MERGE INTO silver.orders AS t
USING staged_changes AS s ON t.id = s.id
WHEN MATCHED AND s.lsn > t._source_lsn THEN
  UPDATE SET status = s.status, total_cents = s.total_cents,
             _source_lsn = s.lsn, _deleted = (s.op = 'd'), _updated_at = now()
WHEN NOT MATCHED THEN
  INSERT (id, status, total_cents, _source_lsn, _deleted, _updated_at)
  VALUES (s.id, s.status, s.total_cents, s.lsn, s.op = 'd', now());

The comparison only works within one source database, because each database has its own WAL. That is fine here: every table comes from exactly one database. staged_changes must also hold only the latest change per id from the batch, because a MERGE may not match the same target row twice.

Run on PostgreSQL 18 for one order: create at LSN 100 gives PENDING; ship at 200 gives SHIPPED; replaying the old create at 100 leaves SHIPPED; a duplicate of the ship changes nothing; a delete at 300 sets _deleted. The same replay without the guard turns the order back into PENDING. A batch with two changes for the same order fails with MERGE command cannot affect row a second time, which is why the batch is reduced to one change per key first.

Bronze is partitioned by ingest date because that is how it is replayed and how retention is applied. Silver is keyed by the business key, and gold by the dimensions people filter on.

The Architecture of Data Motion

At scale, data ingestion moves from bulk nightly queries to real-time streams of events.

Data Motion Architecture from Operational Systems to Data Warehouses

1. Change Data Capture (CDC)

Querying a database for new records (SELECT * FROM orders WHERE updated_at > ?) looks reasonable and quietly loses data. A deleted row has no updated_at to find, so deletes vanish. A transaction that started before your last poll but committed after it carries an older updated_at than your watermark, so its rows are skipped forever. Every row changed twice between polls shows only its final state. And polling a busy table every few seconds adds load to the production database, even with an index on updated_at.

CDC solves this by reading the database's internal Write-Ahead Log (WAL) or binlog. The WAL is a sequential, append-only file the database uses for crash recovery and replication. By tailing the WAL, we get a highly efficient, real-time stream of exactly what changed.

ELI5: Instead of asking the librarian every 5 minutes to check every shelf for new books, we just stand behind the librarian's desk and read their return/checkout logbook over their shoulder.

Anatomy of a CDC Event

A CDC event (like those emitted by Debezium) contains both the before and after state of the row.

// Core CDC Event Structure
type OperationType = 'c' | 'u' | 'd' | 'r'; // create, update, delete, read (snapshot)

interface CDCPayload<T> {
  op: OperationType;
  ts_ms: number; // Timestamp of the change in the source DB
  before: T | null;
  after: T | null;
}

// Example: Order Record
interface OrderRow {
  id: string;
  user_id: string;
  status: 'PENDING' | 'SHIPPED' | 'CANCELLED';
  total_cents: number;
  updated_at: string;
}

// Processing the stream
class OrderCDCProcessor {
  /**
   * Process a CDC event to project the current state into a Data Warehouse.
   */
  public async handleEvent(event: CDCPayload<OrderRow>): Promise<void> {
    switch (event.op) {
      case 'c':
      case 'r':
        // Insert new record
        if (event.after) {
          await this.insertIntoWarehouse(event.after);
        }
        break;
        
      case 'u':
        // Update existing record
        // We have `before` to know what changed, but we upsert the `after` state
        if (event.after) {
          await this.upsertIntoWarehouse(event.after);
        }
        break;
        
      case 'd':
        // Soft or hard delete
        if (event.before) {
          await this.markAsDeleted(event.before.id);
        }
        break;
    }
  }

  private async upsertIntoWarehouse(order: OrderRow) {
    // Upserts must be idempotent. If the same event arrives twice, 
    // the end state in the warehouse must remain correct.
    console.log(`MERGE INTO dim_orders USING ... WHERE id = ${order.id}`);
  }
  
  // Implementation details omitted for brevity
  private async insertIntoWarehouse(order: OrderRow) {}
  private async markAsDeleted(id: string) {}
}

2. Queues and the Transport Layer

Once Debezium extracts changes, it pushes them to a distributed log like Apache Kafka.

Why Kafka?

1. Decoupling: The source DB doesn't care if the data warehouse goes down. Kafka buffers the events.
2. Multiple Consumers: The Fraud team can read the order stream in real-time via Flink, while the Analytics team batch-loads it into Snowflake hourly.
3. Ordering: Kafka guarantees strict ordering within a partition (e.g., partitioning by order_id ensures an update never processes before a create).

3. Storage: Data Lake vs. Data Warehouse

The destination for analytical data usually follows a Medallion Architecture:

• Bronze: Raw data dumped straight from Kafka. Unstructured, append-only.
• Silver: Cleaned, filtered, typed data. Often joined with reference data.
• Gold: Business-level aggregates (e.g., "Daily Sales by Region").

Data Lake (e.g., AWS S3 + Apache Iceberg)

A data lake is cheap object storage containing files (Parquet, ORC). By overlaying a table format like Apache Iceberg, you get ACID transactions on top of S3. You can run massive MapReduce/Spark jobs against it.

Data Warehouse (e.g., Snowflake, BigQuery)

A warehouse provides a SQL interface over columnar storage with an optimized query engine. It is more expensive but provides incredibly fast query performance for business intelligence (BI) tools. Modern systems often blend the two by querying Lake files from the Warehouse ("Data Lakehouse").

4. Backfills and Data Quality

Analytical pipelines break. Schemas change. Logic bugs corrupt downstream aggregates. You will inevitably need to backfill—reprocess historical data with new logic.

How to execute a Backfill

1. Never mutate Gold tables in-place.
2. Spin up a parallel pipeline reading from the Bronze layer (which is immutable and raw).
3. Write the reprocessed data to a parallel set of Silver/Gold tables (e.g., sales_gold_v2).
4. Validate data quality (row counts, sum of totals) against the old tables.
5. Use an atomic view swap to point downstream dashboards to the new tables.

-- Atomic View Swap for Zero-Downtime Backfill Cutover
-- One statement, so readers see either the old view or the new one, never neither.
CREATE OR REPLACE VIEW sales_gold AS SELECT * FROM sales_gold_v2;

Data Quality Checks (Data Contracts)

Data producers (software engineers) often break downstream pipelines by dropping or renaming columns without telling the data team. To prevent this, enforce Data Contracts at the schema registry level. If a PR drops a column that a downstream Flink job relies on, the CI/CD pipeline should fail.

Scale and failure

What breaks	Blast radius	How you find out	What keeps it running	Recovery
Connector stops	source disk fills with WAL	replication-slot lag in bytes	max_slot_wal_keep_size cap	restart; re-snapshot if the cap was hit
Source drops a column	consumers fail or go quietly wrong	contract check at registration	backward-compatibility rule	refuse the change, or version the topic
Older change arrives after a newer one	silver shows stale values	rare; invisible without the guard	_source_lsn guard in the merge	none needed
Same change delivered twice	none	duplicate rate metric	idempotent merge	none needed
Bad logic in a gold table	wrong numbers on dashboards	data-quality checks; row-count drift	backfill to _v2	validate, then swap the view
Personal data lands in bronze	years of exposure in copies	column scanner on new topics	masking at ingest	rewrite affected partitions

Freshness is measured, not assumed. Each silver table records the newest ts_ms it has applied, and an alarm fires when "now minus newest applied change" exceeds 5 minutes. Without it, a stalled pipeline looks exactly like a quiet day.

What the interviewer will push on

1. "How do you handle schema evolution?"Answer: Schema Registry. Avro or Protobuf schemas version the data. When the source DB schema changes, the CDC tool registers a new schema version. Downstream consumers are built to handle forward/backward compatibility (e.g., assigning default values to new columns).
2. "Why use CDC instead of a dual-write from the application?"Answer: Dual-writes (writing to the DB, then firing an event to Kafka from application code) lead to inconsistency. If the DB commits but the Kafka publish fails (or vice versa), the systems drift. CDC relies on the DB's transaction log, guaranteeing that if a commit happens in the DB, it will eventually reach Kafka. Strictly, this is not yet the transactional outbox pattern. CDC on your business tables publishes table rows, so every consumer becomes coupled to your internal schema. The outbox pattern (10.8.4) writes a deliberately designed event row into an outbox table in the same transaction as the business change, and CDC then relays only that table. The strong answer names both and says which one fits: raw table CDC for analytics copies, the outbox for events other services build on.
3. "How do you guarantee exactly-once processing?"Answer: You rarely do. Rely on at-least-once delivery from Kafka, and build idempotent consumers. In a warehouse, use MERGE statements on primary keys so re-processing a row simply overwrites it with the identical state.

Recall

• CDC (Change Data Capture): Tailing the DB write-ahead log to extract row-level changes without impacting OLTP performance.
• Kafka: The central nervous system buffering events and guaranteeing partition-level ordering.
• Medallion Architecture: Bronze (Raw), Silver (Cleaned/Typed), Gold (Business Aggregates).
• Backfills: Reprocessing historical Bronze data into a parallel v2 table, followed by an atomic view swap.
• Data Contracts: Enforcing schema compatibility between producers and consumers to prevent pipeline breakage.

Quiz

1. Why is dual-writing from application code to both a database and a queue an anti-pattern compared to CDC?
2. In CDC, why is it critical that events for the same order_id are routed to the same Kafka partition?
3. What is the purpose of the "Bronze" layer in a Medallion architecture?

Answers

1. Dual-writes lack distributed transactions. A network failure can cause the database write to succeed while the queue write fails, leading to permanent data drift. CDC uses the database's internal transaction log, guaranteeing consistency.
2. Kafka only guarantees ordering within a single partition. If a create and update for the same order go to different partitions, a consumer might process the update before the create, causing an error.
3. The Bronze layer serves as an immutable, raw history of all events. It allows you to replay and backfill data if your downstream processing logic (Silver/Gold) changes or has bugs.