Data Integration Patterns

Data Engineering & Modern Data Platforms · Lesson 12 · Data Ingestion, ETL and Orchestration

Lesson Overview

On the Tuesday after TechLearningPro's enrollment week, a new platform engineer opened four tickets that all said "ingest this." Enrollment lived in PostgreSQL and was the source of record for who may open a course. Progress events, about fifty million on the peak day under the planning assumption from Introduction to Data Engineering, were already durable in that same database because the player writes there first. A customer-success export of assigned enterprise seats arrived as a CSV on a landing prefix, once a night, from a process that will not accept a database connection. The content catalog, roughly eight thousand lessons in the planning assumption, changed through an HTTP API the studio already owned. The proposed design in the ticket was one sentence: put all four on a log and let consumers subscribe.

That sentence is a tool choice wearing the costume of an architecture. The Monday business review still needs yesterday's lesson completions by 06:00 UTC, with a three-day restatement on event time. Instructor operations wants a cohort counter that is wrong if an unenrollment is fifteen minutes late. The seat file's producer has a contract that ends when the object and its manifest exist. The catalog's owners will give a paginated HTTP read and will not give logical-replication credentials. One movement pattern cannot satisfy four source contracts and two clocks. The log is a real option, taught in /learn/kafka and built out in Module 4. It becomes the design when a consumer's freshness bound and the source's ability to emit an ordered change both say so. It is not the design because the other three sources were annoying to schedule.

This lesson is the pattern catalog for moving a fact from a system of record into the analytical estate. A pattern names the trigger, the completeness rule, the ordering assumption, what happens to a delete, and what a replay is allowed to do. The tool comes after the pattern. ETL and ELT Architecture decides where the transform runs. Building Batch Data Pipelines with Python implements one batch shape. API and File-Based Data Ingestion implements pull and file handoff. Change Data Capture Fundamentals is the foundation for reading changes, including deletes, from an operational database. Kafka plus Debezium is lesson 46, not this one.

The running estate is unchanged. The player and the enrollment service write PostgreSQL. The lesson-completion warehouse is derived. It is not allowed to become the resume path. Email, name, and payment tokens are prohibited in the analytical copy. Planning numbers in this lesson are labeled assumptions, carried forward so a design review has something to reject: about two million active learners in a normal month, fifty million progress events on a peak day, a seat file on the order of forty thousand rows a night, a catalog of about eight thousand lessons. If a measurement replaces an assumption, the pattern decision is revisited with the measurement, not with the slogan.

You will leave able to reject "one bus for every source," to write a pattern per data product, and to say what would force the enrollment snapshot to become change data capture. Lesson 13 then places the transform. The bytes from Data Formats and Serialization are what these patterns move.

Learning Objectives

By the end of this lesson you will be able to:

1. Define an integration pattern as a contract about trigger, completeness, order, deletes, and replay, separate from the tool that implements it.
2. Match TechLearningPro sources to full snapshot, incremental watermark, micro-batch, stream, change data capture, API pull, and file drop, including two products on one table.
3. Explain why a watermark on an unmaintained column, and a watermark committed before the load is durable, both lose or double facts.
4. State what at-least-once delivery requires of the destination, and what a dual write does between a commit and a second system.
5. Apply the six-part decision lens so a freshness bound, a delete rule, and operability eliminate patterns before a product name is chosen.
6. Record a pattern decision in a typed review aid and defend the choice for the completion warehouse versus the cohort counter.

Prerequisites

• Lesson 1's lifecycle, grain, and decision lens. Integration is the ingestion stage made specific. A pattern that cannot name the fact, the grain, and the consumer is not finished.
• Lesson 2's separation of operational store and derived analytical estate, and lesson 11's rule that raw bytes keep the producer's format. This lesson does not re-teach Parquet row groups. It does assume you will not convert a partner CSV into a new shape before you have a completeness rule for the file.
• You can read Python 3.11 dataclasses and type hints, as in /learn/python and lessons 4 and 5. The review aid uses the standard library only.
• You can read a parameterized SQL filter. /learn/sql and lesson 6 own query design. Lesson 8's picture of PostgreSQL as the operational engine is enough to understand why an extract wants a snapshot and a replica, not a scan of the primary at noon.
• /learn/kafka is the vocabulary for a log. This lesson uses that vocabulary to decide when a log is the pattern. It does not configure topics, consumer groups, or connectors.

Detailed Core Concepts

A pattern is a contract, and a tool is a mechanism

An integration pattern is the repeatable way a fact leaves its system of record and becomes durable in the analytical estate. The contract has five clauses.

The trigger says what starts a movement: a clock, the appearance of a manifest, a commit in a log, an HTTP schedule. "When we remember" is not a trigger.

Completeness says which source facts are in this movement and which are deliberately outside it. A nightly snapshot is complete for "enrollment as of extract time." It is not complete for "every status the row held at 14:00." An incremental extract is complete for "rows whose watermark falls in the closed-open window," and only if the watermark column actually moves when the business fact moves.

Order says which sequence a consumer may assume. A file has no order except the order you impose by a key. A database snapshot has a transaction snapshot. A log has an offset order that is not the same thing as event time. Lesson 1 already separated those clocks. The pattern must say which one it preserves.

Delete behavior says how a removal in the source becomes visible downstream. Append-only progress events do not delete. A hard delete of a test enrollment disappears from a later snapshot and remains in an earlier snapshot if you kept it. A watermark query that filters updated_at never sees the delete, because the row is gone and cannot satisfy the filter. That single sentence eliminates incremental watermark for any product that must drop a current-state row when the source drops it.

Replay says what a second delivery of the same movement does. Files are retried. Jobs are rerun. Pages of an API are fetched twice after a timeout. The honest delivery assumption for this module is at-least-once. The destination is idempotent, or the pattern is unfinished.

Gregor Hohpe and Bobby Woolf's Enterprise Integration Patterns named the messaging versions of these ideas: document message, event message, polling consumer, competing consumers, idempotent receiver. The book is about application integration. The data-engineering use is narrower and stricter, because the receiver keeps history and a metric. A pattern name without the five clauses is a slide title. Airflow can schedule all of them. A log can carry a nightly file's worth of rows. Neither fact chooses the pattern.

The catalog this course actually uses

Seven patterns cover the sources in this module. Replication is discussed beside them because teams use the word as if it were an analytical product.

Full snapshot batch. On a schedule, copy the current projection of a source into a partition labeled by extract time. Completeness is the snapshot. Intermediate updates between snapshots are not recovered unless an earlier snapshot was retained. Hard deletes are visible as absence when you compare snapshots, and invisible if you only keep the latest copy and upsert blindly. Cost is a full read of the projection each run. For a narrow enrollment projection at the planning volume, a nightly snapshot off a replica is a boring and appropriate design for a next-morning consumer. For the full history of progress events, a nightly recopy of every event ever stored is a scan the primary should never do and a raw zone that cannot be reasoned about.

Incremental batch with a watermark. Copy rows with a watermark strictly after the last committed value, usually with a deliberate overlap. The watermark is a column or a log position that moves forward when a fact is born or changed. Progress events fit when event_time or an insert timestamp is assigned by the source and the product is allowed to restate a short window. The destination replaces a partition or deduplicates on event_id. Hard deletes are out of scope for this pattern. A column named updated_at that the application updates on some writes and not others is not a watermark. It is a rumor.

Micro-batch. The same contract as incremental batch, on a short period: five minutes, fifteen minutes. Freshness improves by the period plus runtime. You pay job startup, source load, and small files. Lesson 11's file-size warning applies directly. A micro-batch that writes one object per run, per hour, per tenant, will present the curated reader with a small-file problem the freshness win does not pay for. Micro-batch is not a stream. It has no continuous offset. It still needs a restatement rule for late event time. It is the right step when a consumer's bound is inside a day and the watermark is real, and when the squad can operate many short jobs.

Stream. A consumer reads an append-only log continuously and records its position. Freshness can be seconds if the producer emits promptly and the consumer keeps up. Operability includes lag, reprocessing, poison messages, and a person who can be woken up. The pattern is justified by a consumer who will take a wrong action on stale data, and by a source that actually emits an ordered record of the fact. Volume alone does not justify it. Fifty million progress events in a day are a large batch. They are a stream only if someone needs them before the batch bound. Module 4 is that design. This lesson's job is to refuse it for the Monday review.

Change data capture. Read inserts, updates, and deletes from a database's change feed. The feed might be the write-ahead log, a trigger-maintained audit table, or, weakest, a query. Lesson 16 compares those mechanisms. At pattern level, CDC is how a current-state table with hard deletes and an intraday bound moves without pretending updated_at is a log. It is a poor default for an append-only event table you can already extract by time. It is the wrong promise if the database owner will not allow a replication slot or an audit table. Lesson 46 implements the log-based path with Kafka and Debezium. Using the product name here would skip the pattern.

API pull. A scheduled client calls an HTTP API, walks pages, and lands them. Completeness is "the pages the API agreed to return for this cursor or this updated-since parameter." Rate limits, partial pages, and a cursor that shifts under offset pagination are the failure modes. Lesson 15 is the implementation. The catalog service is the source that forces this pattern: the analytical estate is not offered a table.

File drop. The producer writes objects and a manifest, or a sibling "done" marker, to a landing area. Completeness is "every part named in the manifest, with the agreed checksum and size." Reading a CSV that is still being uploaded is a torn file, and torn files become wrong metrics with valid-looking rows. The seat assignment export is this pattern. The analytical team does not get to demand a socket to the producer's database because a socket would be more elegant.

Replication, named carefully. Physical replication copies database pages so a standby can take over. It is high availability. Pointing a dashboard at a hot standby still couples analysis to an operational engine and still has no grain. Logical replication copies row changes to another PostgreSQL node. It is adjacent to CDC and is a legitimate way to hold a queryable copy off the primary. A replica used as the extract source is an operational tactic inside snapshot, watermark, or CDC. It is not, by itself, the completion warehouse. Lesson 8 is the internal picture. The integration rule is shorter: the replica protects the primary during a read. The data product still has a grain, a projection, and a prohibited-field list.

Two products, one table

Enrollment is one operational table. It feeds two analytical products with two patterns.

The daily enrollment snapshot serves finance and the completion join. Freshness is the previous day by 06:00 UTC. Grain of the landed snapshot is one row per enrollment key per extract date. Hard-deleted test rows disappear from later dates and remain on earlier dates, which is what a "who was enrolled that morning" question wants. Full snapshot is enough. CDC would also work and would cost a slot, a consumer, and an on-call story the daily question does not use.

The cohort counter serves instructor operations. Freshness is fifteen minutes. The question is "who is enrolled now." A hard unenrollment must disappear. A watermark that misses deletes will keep selling a full cohort. CDC, or a snapshot every few minutes if the table is small enough that a frequent full read is cheaper than operating a change feed, is the pattern. The frequent snapshot is still a snapshot: absence is the delete. It is not a watermark.

Progress events feed the completion warehouse. They are append-only, keyed by event_id, with event time for the business day and ingestion time for pipeline health. The consumer bound is 06:00 UTC with a three-day restatement. Incremental batch with overlap, landing one UTC day per partition, is the pattern. A stream of the same events is a second product, built when a consumer writes down a bound in minutes. Until that sentence exists, the stream is a second system to fail.

The seat file and the catalog API are different sources, not different products on enrollment. They do not get to ride the progress watermark because they do not have one.

Delivery, dual writes, and the outbox

Most movements in this module deliver at least once. The HTTP client retries. The file is copied twice because the job died after the upload and before the checkpoint. The log consumer restarts from the last committed offset and re-reads. Exactly-once is a property you earn with a transaction that commits the offset and the sink write together, in a system that actually offers that transaction. It is not a checkbox on a diagram. Lesson 41 treats Kafka's version. Here, the practical rule is an idempotent landing: replace the partition for that extract, or deduplicate on a producer key, and record how many duplicates you collapsed.

A dual write is the application committing to PostgreSQL and then writing the analytical store or the log as a second step. If the process dies between the two, the source of record has a fact the warehouse will never see, or the retry writes the analytical copy twice. There is no watermark that repairs a fact the application never handed you. The transactional outbox keeps the integration honest: the business transaction inserts the enrollment row and a row in an outbox table in the same commit. A separate publisher reads the outbox and delivers it. The publisher is at-least-once. The outbox row has an identity so delivery is idempotent. CDC from the enrollment table is the other honest option, because the log is written by the same transaction machinery as the row. Both patterns exist so the application does not pretend two systems commit as one. This lesson does not build a publisher. It forbids the dual write as a design you cannot replay.

Notification versus state transfer

A message that says "enrollment 8841 changed" forces every consumer to call back to the source. The callback sees the latest state, not the state at send time, and it puts analytical read load on the operational service. A message or a file row that carries the projected columns the product needs can be replayed from the raw zone without calling back. Progress events in lesson 1 are state-carrying records: event_id, learner_id, lesson_id, event_type, attempt_number, event_time. A webhook that only carries learner_id is a notification. Lesson 15 shows how to verify a webhook. The pattern decision is earlier: if the raw zone cannot rebuild the fact without the source, you integrated a pointer, not a fact.

What integration deliberately stops short of

Integration finishes when the raw zone holds a durable copy with a completeness marker: a committed watermark, a snapshot time, a manifest checksum, or a consumed log position. Validation that rejects a prohibited field can sit on that boundary, because a field that lands in raw storage has already leaked. Business definitions, such as completion version completion-v3, are transforms. Lesson 13 places them. Orchestration, in lesson 19, is what notices the movement did not happen. Mixing those into the pattern discussion is how a review approves Airflow and never writes down deletes.

Architecture and Design Diagrams

The estate has four sources and three analytical products. Arrows are copies. Dotted arrows are refusals. The completion warehouse still must not drive the player.

mermaidCopy
flowchart LR
    subgraph sources [Systems of record]
        pg[(PostgreSQL enrollment and progress)]
        catalog[Catalog HTTP API]
        seats[Seat file producer]
    end

    subgraph raw [Raw zone by pattern]
        snap[Daily enrollment snapshot]
        incr[Progress incremental partition]
        cdc[Intraday enrollment changes]
        api[Catalog page landing]
        file[Seat manifest landing]
    end

    subgraph products [Derived products]
        mart[Completion warehouse]
        cohort[Cohort counter]
        dim[Lesson dimension]
    end

    pg -->|full snapshot nightly| snap
    pg -->|watermark batch| incr
    pg -->|CDC for current state| cdc
    catalog -->|API pull| api
    seats -->|file drop| file

    snap --> mart
    incr --> mart
    cdc --> cohort
    api --> dim
    file --> mart

    mart -.->|must not read| pg
    cohort -.->|must not be called enrollment history| mart

Read the cohort path and the completion path as two products. They both start from enrollment's database and they do not share a freshness contract. Feeding the Monday mart from the fifteen-minute change feed would make a financial-adjacent number depend on a consumer that is allowed to be briefly behind and then caught up. Keeping them separate is the architecture.

A second diagram is the elimination order used in a design review. The order is the point. Freshness and deletes remove options before anyone says "Kafka" or "Airflow."

mermaidCopy
flowchart TD
    start[Name the product, grain, and consumer bound]
    deletes{Must hard deletes disappear from current state?}
    day{Bound is next morning or coarser?}
    snap[Full snapshot retained by extract date]
    feed{Source exposes a change feed you may use?}
    cdc[Change data capture]
    freq[Frequent snapshot if the table is small]
    only{Only an API or only a file?}
    api[API pull]
    file[File drop]
    wm{Watermark column moves on every fact change?}
    inc[Incremental or micro-batch with overlap]
    stream{Wrong action if data is minutes stale, and source emits events?}
    log[Stream, Module 4]
    refuse[Do not invent a feed the source does not have]

    start --> deletes
    deletes -->|yes| day
    day -->|yes| snap
    day -->|no| feed
    feed -->|yes| cdc
    feed -->|no| freq
    deletes -->|no| only
    only -->|API| api
    only -->|file| file
    only -->|query or events| wm
    wm -->|yes| inc
    wm -->|no| stream
    stream -->|yes| log
    stream -->|no| refuse

The third diagram is the watermark lifecycle. The position advances only after the destination acknowledges the load. The overlap is re-read on purpose.

mermaidCopy
sequenceDiagram
    participant Job
    participant Source as Source snapshot
    participant Raw as Raw partition
    participant Mark as Watermark store
    Job->>Mark: Read committed watermark
    Job->>Source: Extract from watermark minus overlap, below upper bound
    Source-->>Job: Rows visible in this snapshot
    Job->>Raw: Replace partition idempotently
    Raw-->>Job: Durable acknowledgment
    Job->>Mark: Commit new watermark
    Note over Job,Mark: A crash before the commit re-reads the overlap

If the job commits the watermark when the extract returns, and dies while writing the partition, the next run starts after the lost rows. The sequence forbids that order. The same picture applies to an API cursor and to a file manifest: record success after the bytes are durable, not when the producer said they exist.

Technical Explanation

Snapshot isolation is part of completeness

A full snapshot that runs SELECT without a transaction snapshot can observe two states of one enrollment. PostgreSQL's REPEATABLE READ or SERIALIZABLE holds a snapshot for the transaction. The extract should be that transaction, off a replica when the planning volume says the primary's noon is too expensive to share. The analytical partition is labeled with the snapshot's timestamp, taken from the database clock inside the transaction, not from the laptop that launched the job. Lesson 8 explains why a single statement can still be a large sequential read. The integration consequence is operational: a snapshot is a read amplification you schedule when the primary can afford it, or you take it on a replica that is known to be caught up within a lag you wrote down.

Comparing two snapshots to find deletes is a join on the enrollment key. Keys in yesterday and missing today are deletions or expirations. Keys in both with changed attributes are updates. Keys only in today are inserts. That diff is only as honest as the projection. If today's extract dropped a column, the diff will claim every row changed. Pin the column list. Lesson 1's prohibited-field rule is a column list with teeth: email in the snapshot fails the run.

Retaining every daily snapshot is a storage cost with a purpose. The completion join for a restated day must use the enrollment picture that the definition names. If the definition says "enrolled as of the end of the event's UTC day," the job reads that day's snapshot, not "whatever enrollment looks like now." Overwriting a single enrollment table in the warehouse destroys that. The pattern of a full snapshot includes the retention of the partitions the restatement window needs, plus whatever finance later requires. Retention is a clause of the pattern, not a cleanup script.

Watermarks, overlap, and the upper bound

Let the committed watermark be the greatest source position successfully loaded. The next extract's lower bound is that value minus an overlap. The overlap exists because clocks skew, because a transaction can commit a row with a timestamp slightly behind the watermark you already stored, and because an in-flight commit may not have been visible in the previous snapshot. The overlap re-reads rows. The destination must treat those rows as already seen.

The upper bound should be the snapshot time, or the snapshot time minus a small safety lag, and it must be decided inside the snapshot. An open-ended WHERE event_time >= lower_bound copies rows that arrive while you read, then a watermark committed at "max event_time seen" races the next insert. A closed-open window [lower_bound, upper_bound) with upper_bound fixed before the scan is a complete extract for that window. Rows with later timestamps wait for the next run, which is what a batch is.

Choose the column with the same care as the grain. For progress events, an insert timestamp assigned by the database when the player commits is a watermark for "what the database accepted." Event time is the business clock for the completion day. They diverge when a phone uploads a checkpoint late. The incremental pattern often watermarks on insert time so the extract does not miss a late upload, and the transform restates event-time days inside the three-day window. Using event time as the only watermark drops the late upload once the window has moved past it, unless the overlap is as large as the lateness you promise to accept. Write that sentence on the product. Lesson 1's restatement window is the consumer-facing half. The watermark column is the extract half. They are not interchangeable.

A tie at the same timestamp needs a secondary order, usually the primary key, and the comparison must be stable. "Greater than max(timestamp)" loses the other rows that shared the max timestamp and were not included because the scan stopped. The overlap hides small ties. It does not hide a watermark that ignores the primary key when thousands of events share one second. Prefer a keyset condition (event_time, event_id) > (:ts, :id) when the volume in one second is large, and still overlap, and still deduplicate.

Idempotent landing shapes

Three destination shapes match the patterns in this lesson.

A partition replace writes the extract for one window to a new prefix and swaps the marker for that window. Re-running replaces. The completion raw zone wants this for one UTC day of progress events. Partial files from a failed attempt must not be in the prefix the transform reads. Write to a temporary prefix and publish the marker last, the same idea as a manifest.

An upsert on the natural key fits a current-state table built from a feed of changes: one row per enrollment key in the cohort store. Re-applying the same change is a no-op if the payload matches. A changed payload for the same key is an update, ordered by source position, not by arrival time at the warehouse. Lesson 16 makes the position a log sequence number. Here, the requirement is that arrival order must not win.

A deduplicating append stores immutable events and collapses event_id at read or at a compacting step. Identical retries collapse. A second payload for the same event_id is a contract breach, as the lesson 1 checker already decided. The integration job counts collapses. A silent DISTINCT in a dashboard is not this shape.

Micro-batch cost, stated with the planning volume

Assume a fifteen-minute micro-batch of progress events on a peak day of fifty million events. That is ninety-six runs and about five hundred thousand events per run. Each run that writes a new file per lesson, and there are thousands of lessons, produces a small-file storm by evening. Each run that writes one compacted file per run produces ninety-six files per day, which a daily reader can handle if the files are large enough to justify a footer and a scan. The arithmetic is the design. "Near real time" without a file-per-run plan is how lesson 11's layout is destroyed by lesson 12's trigger.

Source load is the other cost. Ninety-six indexed reads of "watermark greater than X" are usually cheaper than ninety-six full scans, and more expensive than one nightly read of the day's inserts. The cohort counter's table is small relative to progress events. A full snapshot every fifteen minutes of a narrow enrollment projection can be cheaper and safer than CDC if the replica serves it and the row count stays near the active-enrollment assumption. Measure. The pattern catalog includes that option under "frequent snapshot," and the elimination diagram sends you there when deletes matter and a change feed is unavailable or not worth operating.

Security sits on the copy, not after it

Every pattern copies bytes. The projection is part of the pattern, not a later cleanup. Seat files from customer success often contain a learner email because the upstream process was built for humans. The file-drop pattern for that source includes a reject-or-strip step before the shared raw prefix is visible to analysts. Stripping after analysts have queried the prefix is an incident. API pulls have the same rule: the catalog response may include an author's personal address in a nested field the documentation called optional. Land the projected fields. Lesson 61 is the full privacy playbook. The integration obligation is to refuse the extra field at the boundary you control.

Credentials follow the source. A replication role that can read every table is not the role that extracts enrollment. A token for the catalog API is not stored in the raw manifest. A seat-file location that is world-readable is a second copy of the prohibited column. Pattern reviews that stop at "batch or stream" have skipped the clause that causes the actual incident.

Cost and operability decide ties

When two patterns both meet correctness and freshness, choose the one the current squad can replay on a Sunday. A nightly snapshot plus a daily incremental job is two batch definitions, both rerunnable by re-executing a window. CDC adds slot monitoring, WAL retention when the consumer stalls, and a schema-evolution story for the change envelope. Those costs are justified by the cohort counter's fifteen minutes and its deletes. They are not justified by a desire to retire the snapshot early. Operating both, on purpose, for two products, is cheaper than discovering on a Monday that the single clever feed was down and the snapshot path no longer exists.

Practical Implementation or Design Walkthrough

Problem

Record the pattern for each analytical product, and compute a watermark window that is safe to re-run. The code is a review aid. It encodes the elimination rules from this lesson so a design review can show the clause that fired. It is not a planner you point at production and trust. A human still writes the reason down.

Source requirements

Each request names the product, the source capabilities, the delete behavior, the freshness bound in minutes, and whether a change feed is actually permitted. The four TechLearningPro requests below are the acceptance tests.

Implementation

The module uses the Python 3.11 standard library. No warehouse driver, no log client.

pythonCopy
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from enum import Enum


class Pattern(Enum):
    FULL_SNAPSHOT = "full_snapshot"
    INCREMENTAL_WATERMARK = "incremental_watermark"
    MICRO_BATCH = "micro_batch"
    STREAM = "stream"
    CDC = "cdc"
    API_PULL = "api_pull"
    FILE_DROP = "file_drop"


class DeleteBehavior(Enum):
    APPEND_ONLY = "append_only"
    HARD_DELETE = "hard_delete"


@dataclass(frozen=True)
class IntegrationRequest:
    """One data product against one source. The same table may appear twice."""

    product: str
    source: str
    delivers_files: bool
    exposes_http_only: bool
    has_reliable_watermark: bool
    allows_change_feed: bool
    emits_events: bool
    delete_behavior: DeleteBehavior
    freshness_minutes: int
    daily_rows: int


@dataclass(frozen=True)
class PatternChoice:
    pattern: Pattern
    reason: str


def choose_pattern(request: IntegrationRequest) -> PatternChoice:
    """Apply this lesson's elimination order. Raise if no honest pattern exists."""
    if request.delivers_files:
        return PatternChoice(Pattern.FILE_DROP, "completeness is the producer manifest")
    if request.exposes_http_only:
        return PatternChoice(Pattern.API_PULL, "the service offers pages, not a table or a log")

    hard_delete = request.delete_behavior is DeleteBehavior.HARD_DELETE
    next_morning = request.freshness_minutes >= 12 * 60

    if hard_delete and next_morning:
        return PatternChoice(
            Pattern.FULL_SNAPSHOT,
            "absence in a retained daily snapshot is the delete",
        )
    if hard_delete and request.allows_change_feed:
        return PatternChoice(
            Pattern.CDC,
            "current state must drop hard deletes inside the freshness bound",
        )
    if hard_delete:
        return PatternChoice(
            Pattern.FULL_SNAPSHOT,
            "no change feed; a frequent full snapshot is the delete-safe fallback",
        )
    if request.has_reliable_watermark and request.freshness_minutes <= 30:
        return PatternChoice(
            Pattern.MICRO_BATCH,
            "trusted watermark and a bound inside a day",
        )
    if request.has_reliable_watermark and next_morning:
        return PatternChoice(
            Pattern.INCREMENTAL_WATERMARK,
            "append-only facts, daily bound, overlap plus idempotent partition",
        )
    if (
        request.emits_events
        and request.freshness_minutes <= 5
        and request.daily_rows >= 1_000_000
    ):
        return PatternChoice(
            Pattern.STREAM,
            "minute-level bound on an event source; Module 4 owns the log",
        )
    raise ValueError(f"no pattern for {request.product}; do not invent a feed")


@dataclass(frozen=True)
class Watermark:
    """Committed position. The next read overlaps so late commits are not skipped."""

    committed: datetime
    overlap: timedelta


def extract_window(
    watermark: Watermark,
    upper_bound: datetime,
) -> tuple[datetime, datetime]:
    """Closed-open window [lower, upper). Upper is fixed from the source snapshot."""
    if watermark.overlap < timedelta(0):
        raise ValueError("overlap must be zero or positive")
    lower = watermark.committed - watermark.overlap
    if upper_bound <= lower:
        raise ValueError("upper bound must be after the overlapped lower bound")
    return lower, upper_bound


def commit_watermark(
    previous: Watermark,
    upper_bound: datetime,
    load_durable: bool,
) -> Watermark:
    """Advance only after the partition is durable. Never move backward."""
    if not load_durable:
        return previous
    if upper_bound < previous.committed:
        return previous
    return Watermark(committed=upper_bound, overlap=previous.overlap)


def _demo() -> None:
    progress = IntegrationRequest(
        product="completion_raw_events",
        source="progress_event",
        delivers_files=False,
        exposes_http_only=False,
        has_reliable_watermark=True,
        allows_change_feed=True,
        emits_events=True,
        delete_behavior=DeleteBehavior.APPEND_ONLY,
        freshness_minutes=24 * 60,
        daily_rows=50_000_000,
    )
    daily_enrollment = IntegrationRequest(
        product="enrollment_daily_snapshot",
        source="enrollment",
        delivers_files=False,
        exposes_http_only=False,
        has_reliable_watermark=False,
        allows_change_feed=True,
        emits_events=False,
        delete_behavior=DeleteBehavior.HARD_DELETE,
        freshness_minutes=24 * 60,
        daily_rows=2_000_000,
    )
    cohort = IntegrationRequest(
        product="cohort_counter",
        source="enrollment",
        delivers_files=False,
        exposes_http_only=False,
        has_reliable_watermark=False,
        allows_change_feed=True,
        emits_events=False,
        delete_behavior=DeleteBehavior.HARD_DELETE,
        freshness_minutes=15,
        daily_rows=2_000_000,
    )
    catalog = IntegrationRequest(
        product="lesson_dimension",
        source="catalog_api",
        delivers_files=False,
        exposes_http_only=True,
        has_reliable_watermark=False,
        allows_change_feed=False,
        emits_events=False,
        delete_behavior=DeleteBehavior.HARD_DELETE,
        freshness_minutes=60,
        daily_rows=8_000,
    )
    seats = IntegrationRequest(
        product="assigned_seats",
        source="seat_export",
        delivers_files=True,
        exposes_http_only=False,
        has_reliable_watermark=False,
        allows_change_feed=False,
        emits_events=False,
        delete_behavior=DeleteBehavior.HARD_DELETE,
        freshness_minutes=24 * 60,
        daily_rows=40_000,
    )

    assert choose_pattern(progress).pattern is Pattern.INCREMENTAL_WATERMARK
    assert choose_pattern(daily_enrollment).pattern is Pattern.FULL_SNAPSHOT
    assert choose_pattern(cohort).pattern is Pattern.CDC
    assert choose_pattern(catalog).pattern is Pattern.API_PULL
    assert choose_pattern(seats).pattern is Pattern.FILE_DROP

    start = datetime(2026, 3, 2, 0, 0, tzinfo=timezone.utc)
    mark = Watermark(committed=start, overlap=timedelta(minutes=10))
    upper = datetime(2026, 3, 3, 0, 0, tzinfo=timezone.utc)
    lower, end = extract_window(mark, upper)
    assert lower == start - timedelta(minutes=10)
    assert end == upper
    stalled = commit_watermark(mark, upper, load_durable=False)
    assert stalled == mark
    advanced = commit_watermark(mark, upper, load_durable=True)
    assert advanced.committed == upper
    print(
        f"progress={choose_pattern(progress).pattern.value} "
        f"cohort={choose_pattern(cohort).pattern.value} "
        f"window={lower.isoformat()}..{end.isoformat()}"
    )


if __name__ == "__main__":
    _demo()

The SQL that belongs with the progress choice is ordinary parameterized SQL. Bind lower_bound and upper_bound from extract_window. Run it in a snapshot transaction on a replica. The column list is the contract.

sqlCopy
SELECT event_id,
       learner_id,
       lesson_id,
       event_type,
       attempt_number,
       event_time
FROM progress_event
WHERE event_time >= %(lower_bound)s
  AND event_time < %(upper_bound)s;

If the true watermark is an insert timestamp, put that column in the filter and keep event_time in the projection for the transform. Do not filter on a column you do not also store.

What the demo asserts

Identical configuration of enrollment with two freshness values yields a snapshot for the daily product and CDC for the cohort product. A failed load does not move the watermark. The overlap shifts the lower bound backward by ten minutes in the example. Ten minutes is an illustration, not a capacity study. The overlap has to cover the lateness you measured between commit visibility and the watermark value.

How you would run it

bashCopy
python3 integration_patterns.py

Expected output is one line naming incremental_watermark, cdc, and a window from 2026-03-01T23:50:00+00:00 through 2026-03-03T00:00:00+00:00. An assertion failure means the elimination rules and the demo drifted apart.

What this walkthrough does not do

It does not open a replication slot, call an HTTP API, or write Parquet. Those are lessons 16, 15, and 11. It does not schedule itself. Lesson 19 does that. The watermark store in production is a table with the product name, the committed position, and the run id. A JSON file on a laptop is the demo's idea of that table, not the table.

Enterprise Real-World Scenarios

Four sources, one rejected bus

The analytics squad wrote the pattern record after the "put it on a log" review. Assumptions are labeled.

Security is the same on every row of the table: the progress projection has no email, the enrollment snapshot has learner_id and course identifiers and no name, the seat file is rejected if a header named email appears, and the catalog pull drops author contact fields. The completion mart joins the snapshot and the events. The cohort counter is a different serving path. Instructor studio is told, in writing, which one answers "who is in the room now."

The evidence the review asked for, one month later: each product has a committed position or manifest id, a rerun of one progress day does not change the distinct event_id count, and the cohort store's row count drops in a test when a hard delete is applied on the source. A green scheduler with one topic and four unhappy consumers is not evidence.

The hour the cohort counter borrowed the mart

A launch-day incident made the cost of one pattern concrete. Instructor operations asked to "just read enrollment from the completion warehouse" for the live cohort, because the CDC consumer was not built yet. The mart's enrollment picture was the 06:00 snapshot. Unenrollments during the day were invisible. The room overfilled relative to paid seats. The repair was not a faster transform of the mart. The mart's contract stayed next-morning. The cohort product was either a fifteen-minute snapshot of the narrow projection, acceptable at that volume while CDC was built, or an explicit "unavailable" in the UI. Serving a stale contract under a fresh label was the defect.

The second clause of the repair: progress ingestion stayed on the nightly incremental path. Someone had suggested micro-batching all fifty million events "while we are in there." That would have multiplied job overhead and file counts to fix a product that does not read progress events. Pattern reviews fail when a crisis on one product changes the contract of another.

A planning note belongs on the cohort choice. If active enrollment keys are a few million narrow rows, a fifteen-minute snapshot off a replica can meet the bound with less machinery than a logical slot. The review aid returns CDC when a change feed is allowed, because the request said the feed was allowed. The human note can still say "frequent snapshot until the slot is operated." The aid is not allowed to hide that the fallback exists. Write the fallback next to the choice.

Best Practices

• Write the five clauses (trigger, completeness, order, deletes, replay) on the pattern before you name a product or a scheduler.
• Choose the pattern per data product. Two freshness bounds on one table are two patterns.
• Extract from a transaction snapshot, off a replica when the primary's latency budget cannot absorb the scan.
• Pin the column list. Fail the run when a prohibited field appears.
• Advance a watermark, a cursor, or a manifest acknowledgment only after the destination is durable.
• Overlap incremental extracts and make the landing idempotent. Count collapsed duplicates.
• Use insert time or a log position to find rows, and event time to define the business day, when those clocks differ.
• Keep daily snapshots for as long as restatement and finance need "as of that morning."
• Prefer a frequent snapshot over a homemade change feed when deletes matter, the table is small, and nobody will operate a slot.
• Refuse a dual write. Use an outbox or CDC when the application transaction must be the source of the analytical fact.
• Carry state in the raw record when the product must replay without calling the source.
• Revisit the pattern when a measured volume or a new consumer bound contradicts the assumption that justified it.

Common Mistakes and Anti-Patterns

One bus for every source. A log subscribes consumers who do not all share a bound, a delete rule, or an owner. The seat-file producer cannot publish to it, the catalog team will not, and the Monday job gains a dependency it does not use. The repair is the table in the first scenario: a pattern per product, a log only where the elimination order lands on a stream.

Full snapshot of an append-only history. Recopying every progress event each night is simple to explain and hostile to the primary, the network, and the raw zone. The repair is an incremental window on a real watermark, with partition replace for that window.

A watermark the application does not maintain. updated_at defaults to insert time and is not touched when support corrects a status. The incremental job loads a quiet lie. The repair is either a column the transaction updates on every business change, a snapshot, or a log. Test the column by performing the update you care about and showing the watermark moved.

Committing the position before the write. The job records "we are done through Tuesday" when the extract returns, then fails while uploading. Wednesday's job starts at Wednesday. Tuesday is gone from the raw zone and present in the source, which is the worst combination, because a row-count check against "what we loaded" still looks internally consistent. The repair is the sequence diagram: durable landing, then commit.

Dual write from the request handler. The enrollment API writes PostgreSQL, then calls the warehouse loader. A timeout creates a gap or a duplicate. Retries create more duplicates. The repair is an outbox in the same transaction or CDC on the table. The handler returns when the source of record commits.

Reading the hot standby as if it were the mart. Analysts are given the replica to "avoid a pipeline." They join operational tables at operational grain, including columns the mart prohibits, and they disagree with Monday's number. The repair is to use the replica as the extract source for a contracted snapshot, and to remove analytical credentials from it.

Notification without state. A webhook stores learner_id and fetches the profile later. The profile has changed, and the raw zone cannot rebuild Tuesday. The repair is to persist the projected payload, or to stop claiming the webhook is the integration.

Micro-batch without a file plan. Every fifteen minutes writes thousands of KB-scale objects. The morning query spends its time opening files. The repair is one object per run, or a daily compaction with a freshness bound that still holds, or a return to a single daily batch if the consumer never needed fifteen minutes.

Troubleshooting and Design Review

Symptom: Tuesday's progress partition is missing events that exist in PostgreSQL, and Wednesday's partition looks complete.

Likely cause: the watermark advanced past Tuesday before the partition was durable, or the upper bound was "max timestamp seen" during a scan that raced commits, or the filter used event time and the rows were inserted later with an earlier event time.

Investigation: compare source counts for the window under a repeatable-read snapshot with raw counts for that partition. Read the watermark history, including whether a run committed without a durable marker. Check insert time against event time for the missing ids.

Corrective action: stop advancing the watermark on extract. Reload the window with overlap. If event time was the watermark, switch the extract clock to insert time and keep the three-day event-time restatement.

Symptom: The cohort counter still shows a learner who unenrolled an hour ago.

Likely cause: the counter is fed by an updated_at extract, and unenroll is a hard delete. Or it reads the daily snapshot. Or CDC is applied out of order and an older update lands after the delete.

Investigation: look up the key in the source, in the latest snapshot, and in the change feed. Confirm whether any event for the delete exists. Check the position used to order applies.

Corrective action: stop using a watermark for hard deletes. Serve the snapshot whose time you are willing to advertise, or apply the change feed in source order. Do not patch the counter by hand.

Symptom: Seat-file row counts fluctuate by a few percent every night and finance sees seats appear and vanish.

Likely cause: the job read the CSV while the producer was still writing, or it loaded a partial multipart object because no manifest was required.

Investigation: compare landing object size and checksum with the producer's manifest. Look at file modification time versus job start.

Corrective action: require the manifest, load only named parts, and reject a run whose checksums differ. Keep the previous good partition served.

Symptom: The primary's latency climbs at 01:00 UTC.

Likely cause: the snapshot or the incremental extract reads the primary. Peak is not only "daytime" if enrollment week has global learners. Fifty million events is enough to matter.

Investigation: identify the session, the replica lag, and whether the extract could run on the standby. Confirm the standby is within the lag the completeness clause allows.

Corrective action: move the read to the replica. If the replica lags beyond the clause, the pattern's completeness is the replica's lag, and the job should wait or fail visibly rather than read the primary "just this once."

Symptom: Two teams announce different enrollment totals for the same morning.

Likely cause: one team counted the snapshot partition for that extract date, the other counted the cohort table, which is current state and has already applied later unenrollments.

Investigation: compare grain and as-of time, not dashboards.

Corrective action: name the products differently in the serving layer. The cohort table is not "enrollment." The snapshot partition is not "current cohort."

A design review asks, in order: What is the product, the grain, and the bound? What is the source allowed to provide, including deletes and credentials? Which pattern survives the elimination diagram? What is the completeness marker, and when does it advance? What does a second delivery do? Which fields never copy? Who is on call when the marker stalls? A review that starts with a connector has skipped the product.

Interview Questions and Answers

Q1. What is a data integration pattern, and why is "we will use Airflow" not an answer?

A pattern is the contract for moving a fact: what triggers the movement, which source facts count as complete, what order is preserved, how a delete appears, and what a replay does. Airflow can schedule a snapshot, a watermark extract, an API pull, or a file watcher. The scheduler does not decide whether a hard delete is visible. I would answer with the five clauses for the specific product, then name the mechanism. A tool without the clauses is a plan to discover the delete bug in production.

Q2. When would you choose a full snapshot over an incremental extract?

When the consumer needs a picture of current state at a point in time, the source is small enough to reread on the agreed schedule, or hard deletes must show up as absence and there is no reliable change feed. TechLearningPro's daily enrollment snapshot is that case: the completion join wants "enrolled as of that extract," test enrollments are hard-deleted, and the morning bound does not need a log. I would not snapshot the full history of an append-only event table. I would also retain snapshot partitions for the restatement window, because replacing a single current table erases the as-of picture.

Q3. How does a watermark lose rows even when the job is green?

Three common ways. The position is committed before the load is durable, so a crash skips the failed window. The predicate is column > max_seen and several rows share that maximum, so the ones not scanned are never less than the new watermark and never greater either. The column does not move when the business update happens, so the row never re-enters the extract. The repair is a closed window decided inside a snapshot, an overlap, a commit after durability, a tie-breaker key, and a column you have tested against the update you care about. The destination still deduplicates, because the overlap re-reads on purpose.

Q4. What is the difference between micro-batch and streaming?

Micro-batch reruns an incremental contract on a short timer. It has jobs, windows, and files. Streaming consumes a log continuously and tracks a position. Both can land data every few minutes. They do not fail the same way. A micro-batch failure is a run you re-execute. A stream failure is lag, a stuck partition, and a consumer you operate overnight. I pick micro-batch when the bound is short and a watermark is honest. I pick a stream when a minute of staleness causes a wrong action and the source emits events. I do not pick a stream because the daily volume is large. Fifty million events can be one well-partitioned batch.

Q5. When is change data capture the pattern, rather than a better watermark?

When the product is current state, rows are hard-deleted, and the freshness bound is inside the time a full snapshot can comfortably run, and the database owner allows a change feed. A watermark cannot see a row that no longer exists. CDC, or a frequent snapshot if the table is small, can. I would not use CDC as the default for append-only progress events that already have a timestamp and an event_id. Lesson 16 is the mechanism. Lesson 46 is the Kafka and Debezium implementation. Naming the product in the pattern interview is how people skip replica identity and slot retention.

Q6. How do you know a file drop is complete?

The producer publishes a manifest that names every part, with size and a checksum, after those parts are fully visible. The consumer loads only that set, verifies the bytes, and only then marks the batch complete. A CSV that is growing, a multipart upload without a completed marker, or a directory listing used as the contract will ingest a torn file that still parses. The previous complete manifest stays the served copy until the new one verifies. Row count alone is a weak completeness check, because a torn file can have a plausible number of lines.

Q7. Why is a read replica not an analytical data product?

A replica is an operational copy kept for availability and for offloading reads. It has the source grain, the source columns, and the source's availability contract. The completion warehouse is derived, projected, and allowed to be hours behind on purpose. Querying the replica for the Monday metric couples the business number to operational schema changes and invites prohibited columns. I use the replica as the place the snapshot or watermark extract runs, with a lag limit written into completeness. The product is still the partitioned, projected landing.

Q8. What is wrong with writing to the database and to the warehouse from the application handler?

Those are two commits. A failure between them drops the analytical fact or, on retry, duplicates it. There is no source watermark that can reconstruct a write the handler never recorded. I want one transaction that commits the business row and an outbox row, with a publisher that delivers at least once and an idempotent consumer. CDC on the business table is the other design, because the database log is produced by the same transaction. I would not accept "we retry the HTTP call" as a reconciliation story.

Q9. A stakeholder wants progress events on a log so the completion metric updates every minute. What do you ask?

Who takes a different action at minute one than at 06:00 UTC, and what do they do when the log is thirty minutes behind? If the answer is the Monday review, the incremental batch already meets the contract, and a log adds operability without a decision. If the answer is a learner-facing surface, I ask whether that surface is allowed to read a derived store at all. The player is not. I also ask whether late event time is still restated for three days, because a minute-level view that ignores lateness will not match the official day. Module 4 is where we build the log after those answers are written down.

Q10. How should two consumers of the enrollment table be integrated?

As two products. The morning completion join gets a retained daily snapshot. The fifteen-minute cohort counter gets CDC or a frequent snapshot, and it is allowed to show current state, including deletions that the historical snapshot must still contain. I do not point the cohort UI at the mart to save a pipeline, and I do not feed the mart from the intraday feed so that a financial-adjacent number depends on the cohort consumer. Shared source, separate completeness, separate names in the serving layer.

Q11. What does at-least-once delivery require of an integration you operate?

A replay of the same window must not change the metric. Practically, that is partition replace, an upsert ordered by source position, or deduplication on a producer identifier, plus a count of how many duplicates were collapsed. I assume files, API pages, and job retries are delivered more than once, because that is how retries work. I treat exactly-once as a specific transactional claim I can point at, not as a default. If I cannot describe the idempotent key, I do not have a pattern yet.

Q12. What evidence would make you replace the daily enrollment snapshot with CDC?

A consumer whose bound is inside the snapshot schedule, who needs every intermediate update or intraday delete, and who has an owner for slot health and schema change. I would also want a measured cost showing the frequent snapshot no longer fits the replica. I would not replace it because CDC is more modern, and I would not remove the snapshot if any consumer still needs "as of that morning" after later updates have been applied to current state. Adding CDC beside the snapshot is the usual outcome. Deleting the snapshot is a separate decision.

Hands-On Exercises

Exercise: Choose patterns for a learning platform's sources

Problem statement. You are handing Module 2 a pattern record for a learning platform with the same shape as TechLearningPro. Use your own volumes and bounds. The point is the clauses, not the numbers in this lesson.

Tasks.

1. List four sources: an operational enrollment table, an append-only progress table, an HTTP catalog, and a nightly partner file. For each, state what the producer will actually let you do. Mark one capability you are not allowed to assume.
2. Define two products on the enrollment table with different freshness bounds. Write the five pattern clauses for each.
3. Choose a pattern for progress events. Name the watermark column, the overlap, the upper-bound rule, and the idempotent landing shape. Say what happens to a row inserted today with yesterday's event time.
4. Specify completeness for the partner file, including the manifest fields and the prohibited header that fails the run.
5. Apply the six-part lens in a short paragraph per product. Label every volume as an assumption.
6. Implement choose_pattern and commit_watermark from this lesson, or your own equivalents, and add one request that must raise because no honest feed exists.
7. Draw the elimination diagram with your two enrollment products on it, including one "must not" arrow between products.
8. Describe one failure: a watermark committed early, a torn file, or a dual write. Say what the consumer sees and how replay works.

Expected deliverables. A pattern record of at most four pages, the diagram, the runnable checks, and the failure note.

Evaluation rubric.

You pass when a reviewer can point to the completeness marker for each product and to the clause that forbids feeding the cohort counter from the daily mart.

Lesson Summary

An integration pattern is the contract for a movement: trigger, completeness, order, deletes, and replay. The tool implements the contract. It does not choose it. TechLearningPro's progress events, daily enrollment, cohort enrollment, catalog, and seat file are five products across four sources, and they do not share a bus just because a bus was proposed.

Full snapshot retains an as-of picture and shows hard deletes as absence. Incremental watermark and micro-batch move append-only or honestly updated rows, with overlap and a position that advances only after the landing is durable. A stream is a freshness decision with an on-call cost, owned by later lessons. CDC is the pattern when current state must observe deletes inside a short bound and a change feed is permitted. API pull and file drop exist because some sources offer nothing else. Dual writes are refused. Replicas are extract locations, not marts.

Lesson 13 takes these landings and decides where the transform runs: before the raw zone is shared, or inside the analytical store after an immutable raw copy. The patterns in this lesson are finished only when that raw copy has a completeness marker a later transform can trust.

Further Reading

• Gregor Hohpe and Bobby Woolf, Enterprise Integration Patterns — document messages, event messages, polling consumers, and the idempotent receiver. Read it as the messaging ancestor of the five clauses, not as a warehouse design.
• Martin Kleppmann, Designing Data-Intensive Applications — replication, partitions, and batch versus stream. The dual-write failure is the derived-data argument in that book.
• Joe Reis and Matt Housley, Fundamentals of Data Engineering — ingestion as a lifecycle stage, and the warning against choosing a tool before the source and the consumer are named.
• PostgreSQL documentation: transaction isolation — the snapshot a full extract needs. /learn/postgresql and lesson 8 are the course path.
• PostgreSQL documentation: logical replication — the change-feed idea this lesson only selects. Lesson 16 explains it. Lesson 46 implements it with Kafka and Debezium.
• Python dataclasses — the review aid. Language primer: /learn/python.
• Continue this course: ETL and ELT Architecture, then Building Batch Data Pipelines with Python.