Design a data lake ingestion layer

01The design, in the six-step framework

1. Requirements

Functional.

• Land three kinds of data into the lake: application events (from a message log), change-data-capture streams from operational databases (every insert, update and delete), and periodic full dumps from partner systems.
• Data arrives in the lake as queryable tables with schemas, partitioned by time, readable by SQL engines and batch jobs.
• Every row lands exactly once, even when a source or the ingest job retries; late data is folded in.
• Schema changes in the source are detected and applied without breaking readers.
• Out of scope: the query engines, downstream transformations, access control policy beyond table-level grants.

Non-functional.

• Volume: 20 TB per day across 2,000 tables; the biggest event stream is 500,000 messages per second.
• Freshness: event tables queryable within 15 minutes; CDC tables within 5 minutes; dumps within an hour of arrival.
• Durability: no acknowledged data lost; a lineage record for every file saying where it came from.
• Cost: object storage, columnar files, no always-on database for the data itself.

2. Core entities

• Source: id, kind (stream, cdc, dump), connection, schema registry reference, owner.
• Table: name, current schema version, partitioning (by date, hour), format, location.
• Snapshot / commit: table, version, list of added files, list of removed files, committed_at; the unit of atomic visibility.
• Data file: path, table, partition, row count, min/max per column, source offsets covered, checksum.
• Checkpoint: source, partition, last committed offset or LSN, the ingest job's progress.

3. API

• POST /sources {kind, connection, table, partitioning} → registers an ingest; the platform creates the table and starts the job.
• GET /tables/{name}/snapshots → versions; GET /tables/{name}?as_of= → the file list for time travel. Used by query engines through the catalogue.
• POST /tables/{name}/commit {add_files, remove_files, expected_version} → new version or a conflict; the only way files become visible.
• GET /lineage/{file} → source, offsets, job run, schema version.
• POST /schemas/{source} {new fields} → validated for compatibility, assigned a version.

4. High-level design

An event stream landing, step by step:

1. The streaming writer consumes a topic in micro-batches of about 5 minutes, validates each message against the registered schema, and writes columnar files to a staging prefix in object storage, one per partition per batch.
2. When the batch's files are fully written, the writer makes one atomic commit to the catalogue: "table T, version n+1 = version n plus these files, covering offsets a..b", and stores the offsets in the commit itself.
3. Only after the commit succeeds does the writer advance its checkpoint; readers see version n+1 all at once or not at all.
4. Compaction later rewrites many small files into 256 MB files in a new commit; retention drops partitions older than policy in another.
5. Query engines ask the catalogue for the current (or an older) version's file list and read directly from object storage.

5. Deep dives

How is "exactly once" achieved on top of storage that only offers "write a file"?

By making the commit the unit of truth and making it idempotent. Files are written to unique staging paths; writing them twice is harmless because nothing references them until a commit does. The commit records the source offsets it covers; if the writer crashes after writing files but before committing, the restarted writer re-reads from its last committed offset, writes new files and commits those; the orphaned staging files are garbage-collected because no snapshot points at them. If it crashes after committing but before checkpointing, on restart it reads the table's last commit, sees offsets a..b are already covered, and skips them. The condition to state out loud: offsets must be recorded in the same atomic commit as the files, which is why the catalogue, not the storage bucket, is the transactional component.

How do updates and deletes from a database become rows in immutable files?

Change-data-capture gives a stream of (key, operation, before, after, LSN). The simplest landing is append-only: every change becomes a row with its operation and LSN, and readers reconstruct the current state with a window function (latest LSN per key). That is cheap to ingest and slow to read. The alternative is a merge-on-write table: the applier buffers five minutes of changes, then rewrites the affected partitions' files with the changes applied (upsert and delete) in one commit. Reads are fast; writes cost a rewrite of touched partitions, so partition by something updates cluster on (recent dates). Most lakes keep both: a raw change log table for audit and replay, and a merged current-state table for analysts, and the applier is what maintains the second from the first.

How do schema changes avoid breaking readers?

Every file carries its schema; the table's schema is the union of versions with explicit evolution rules: adding a nullable column is allowed, widening a type is allowed, renaming is a new column plus a deprecation, dropping is a metadata-only hide. The registry validates a proposed change before the ingest job adopts it; a source that starts sending an incompatible shape goes to a quarantine table and pages the owner. Readers resolve columns by id, not by position, so old files and new files read correctly in one query. Say "schema on write with evolution rules", because "schema on read" is how lakes become swamps.

Why does the catalogue matter so much, and how does it scale?

Because every read starts with "which files make up version n of table T", and with tens of millions of files a listing of the bucket is both slow and inconsistent. The catalogue stores snapshots as manifests (lists of files with per-file statistics) in object storage, with a small transactional pointer per table in a database; a commit is a compare-and-swap on that pointer. Query planning reads manifests and prunes partitions and files by min/max stats before touching data. This is the design of the open table formats, and naming one (Iceberg, Delta, Hudi) with the reason is fine; reciting their feature lists is not.

6. What a strong answer sounds like

• Frame the lake as immutable files plus a transactional catalogue, and put exactly-once in the commit, not in the file writes.
• Treat the three source kinds differently: micro-batch for events, merge for CDC, load for dumps, with the same commit protocol underneath.
• Count the files and explain compaction; small files are the operational failure mode of every lake.
• Describe schema evolution rules and the quarantine path.
• Expected follow-ups: GDPR deletes (rewrite files containing the key, in a commit); backfills (a new ingest run writing to a branch, then swapped in); cost controls (tiering old partitions to colder storage).
• Common mistake: streaming rows one at a time into tiny files, or trusting the storage listing as the source of truth.

02Where this question was reported

• Blind · Data lake ingestion layer