Streaming & CDC Ingestion

Batch ingestion answers the question “what did this vendor send us overnight?”; streaming ingestion answers a sharper one — “what changed in the source system in the last few seconds, and can we accrue against it before the promotion period closes?” Within the broader data ingestion normalization pipelines discipline, this topic area owns the near-real-time path: capturing row-level changes from POS, ERP, and distributor databases as an ordered stream of change events, and landing them as reconciliation-grade canonical facts without waiting for a nightly file. It exists alongside — not instead of — the file-based lane. Where csv & edi parsing workflows turn a discrete drop into typed records, change data capture (CDC) turns a continuously mutating source table into a durable log of create, update, and delete operations that the reconciliation engine can replay, deduplicate, and reason over deterministically.

This is the canonical reference for that streaming stage. It describes the change-event schema a CDC connector emits, how delivery semantics and per-key ordering are guaranteed, how idempotent upserts keyed on a deterministic hash make replays and duplicates no-ops, how monetary precision survives a serialization boundary that would otherwise destroy it, and how watermarks, offset lag, and dead-letter routing keep a live stream trustworthy. The audience is concrete: Python ETL developers who own the stream consumer, trade finance analysts who must defend a figure that was accrued minutes after a sale, and vendor managers who answer for a disputed deduction. The complement to this page is the batch path handled by async batch processing, and the two share the same canonical fact schema so downstream rating cannot tell — and must not care — which lane a record arrived on.

Positioning Within the Reconciliation Architecture

The streaming lane sits at the front of the pipeline alongside the file-based one, but its contract with the source is fundamentally different. A batch parser reads an artefact the vendor produced deliberately; a CDC connector reads the source database’s own write-ahead log — the same log the database uses to recover after a crash — and reconstructs every committed INSERT, UPDATE, and DELETE as a change event. Log-based capture (the approach popularised by connectors such as Debezium) is preferred over query-based polling because it observes deletes, captures intermediate states, and imposes no incremental-timestamp discipline on tables the reconciliation team does not own. The connector records a monotonic position — a log sequence number or offset — with every event, and that position is the anchor for exactly-once processing and for replay.

Two design commitments keep the lane reconciliation-grade. First, the event log is the source of truth, not the consumer’s state: change events are published to a durable, append-only, partitioned log (a Kafka-style topic) and are never mutated, so a consumer that falls behind, crashes, or ships a bug can rewind to a committed offset and reprocess deterministically. Second, the consumer emits the same canonical facts the batch lane does — it validates and normalizes but does not adjudicate. Deciding whether a captured sale falls inside a promotion window belongs to the eligibility rule framework and, more broadly, to the core architecture / promotion mapping layer that this stream feeds; the streaming stage only guarantees that a change in the source becomes a typed, hashable, correctly-ordered fact with minimal latency. That clean separation is what lets pos & erp sync patterns and the accrual engine treat a streamed record and a batched record as the same thing.

Entity Topology and Schema Specification

A change stream is only replay-safe if every event is self-describing and independently addressable. Each event carries the source position that produced it, the operation, the row image before and after the change, and enough provenance to route, deduplicate, and audit it without consulting any external state. The table below specifies the minimum change-event schema the connector emits and the consumer validates; treat it as a versioned interface where adding a nullable field is backward-compatible and renaming or retyping one is a breaking change registered in the schema registry.

The record_hash is the linchpin of correctness. It is a deterministic digest over the canonical business fields of the after image (source key, business date, SKU, quantity, and the string form of the amount) — never over volatile fields like ingest_ts or event_id. Because it is deterministic, the same logical change always hashes to the same key, so an upsert keyed on record_hash collapses duplicates and replays into a single row. Monetary values travel as strings and are parsed straight into decimal.Decimal, so serialization across the log never routes an amount through a binary float.

Conditional Logic and Rule Integration

Before any fact is written, each event passes a deterministic gate that decides delivery semantics, ordering, and admissibility. These are not implementation details — they are the difference between a stream you can settle against and one you cannot.

Delivery semantics. A durable log gives you at-least-once delivery cheaply: on a consumer restart or rebalance, the last uncommitted batch is redelivered, so a naive consumer double-counts. The streaming lane therefore treats every event as potentially duplicated and achieves effectively exactly-once not by trusting the broker but by making the write idempotent — an upsert keyed on record_hash. This is strictly stronger than broker-side exactly-once alone, because it also absorbs duplicates introduced by connector restarts, source-side retries, and manual replays. The rule is simple: duplicates must be no-ops, and the system must be correct even if every event is delivered twice.

Ordering per key. Reconciliation is order-sensitive within a single row’s history — an update that lowers a rebate amount must not be overtaken by the create it supersedes. Global ordering across a whole topic is neither necessary nor achievable at scale, so the connector partitions by the source primary key, which guarantees ordering within a key while allowing parallelism across keys. The consumer additionally guards against regressions by refusing to apply an event whose source_offset is below the last one already committed for that key.

Tombstones and deletes. A d event carries a populated before image and a null after. The consumer translates it into a tombstone: the canonical fact is soft-deleted (or a compacting tombstone is written to the log) so a captured deletion at the source propagates as a reversal rather than a stale row that keeps accruing. Deletes are as financially significant as creates — a voided POS transaction that never reverses is a phantom accrual.

{
  "delivery": {
    "semantics": "effectively_exactly_once",
    "idempotency_key": "record_hash",
    "on_duplicate": "no_op_upsert",
    "ordering": "per_partition_key",
    "reject_if": "source_offset <= last_committed_offset[key]"
  },
  "op_handling": {"c": "insert_or_upsert", "u": "upsert", "d": "tombstone_soft_delete"}
}

Financial Settlement Layer

Streaming does not rate volume, but the serialization boundary between source, log, and consumer is exactly where monetary precision is silently destroyed if the schema is careless. A JSON number is an IEEE-754 double; encoding 12034.55 as a bare number and decoding it in a consumer can reintroduce the classic 0.1 + 0.2 drift that a batch pipeline works hard to avoid. The streaming lane forbids this categorically: every amount is serialized as a string in the change event and parsed into decimal.Decimal on the way out, honouring the same reproducible-to-the-cent discipline that IFRS 15 / ASC 606 accruals require. Currency is carried as the as-reported ISO 4217 code and never converted inside the consumer, so the streaming path reasons in a single faithful representation of the source and defers FX to the downstream settlement layer.

Quantities follow the same rule and retain their original unit of measure, so a later case-versus-each conversion stays auditable. The extraction idiom below is the only acceptable one — string in, Decimal out, with an explicit guard that rejects any amount that arrived as a float because a producer serialized it wrongly.

from decimal import Decimal, ROUND_HALF_UP

def amount_from_event(raw: object) -> Decimal:
    """Amounts must cross the log as strings. A float here means a broken producer."""
    if isinstance(raw, float):
        raise ValueError("amount arrived as float — precision already lost upstream")
    value = Decimal(str(raw))                       # str() of a str is a no-op; Decimal(str) is exact
    return value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)

Because the write is an idempotent upsert on record_hash, applying the same event twice cannot double an accrual, and because amounts are exact Decimal, a rewound offset that reprocesses a full day of change events reproduces the identical settlement figure byte-for-byte — the property that makes a live stream defensible in an audit.

ETL Implementation Patterns

Enforcement begins at the consumer boundary. Model the change event with Pydantic v2 so a float amount, a missing offset, or a create event that somehow carries a before image fails fast at ingestion instead of surfacing as a bad accrual weeks later. The model below encodes the operation-specific invariants and computes the deterministic record_hash the upsert depends on.

import hashlib
from decimal import Decimal
from enum import Enum
from pydantic import BaseModel, field_validator, model_validator

class Op(str, Enum):
    create = "c"
    update = "u"
    delete = "d"

class ChangeEvent(BaseModel):
    event_id: str
    source_offset: str
    op: Op
    key: str
    after: dict | None = None
    before: dict | None = None
    amount: Decimal
    currency: str
    source_ts: str

    @field_validator("amount", mode="before")
    @classmethod
    def _no_float_money(cls, v: object) -> Decimal:
        if isinstance(v, float):
            raise ValueError("amount must arrive as str or Decimal, never float")
        return Decimal(str(v))

    @model_validator(mode="after")
    def _op_image_consistency(self) -> "ChangeEvent":
        if self.op is Op.create and self.before is not None:
            raise ValueError("create event must not carry a before image")
        if self.op is Op.delete and self.after is not None:
            raise ValueError("delete event must not carry an after image")
        return self

    def record_hash(self) -> str:
        img = self.after or self.before or {}
        canonical = "|".join([
            self.key,
            str(img.get("business_date", "")),
            str(img.get("sku", "")),
            str(img.get("quantity", "")),
            str(self.amount),        # Decimal -> exact string form
        ])
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

Persistence uses idempotent upserts keyed on record_hash, so a redelivered batch after a consumer rebalance overwrites cleanly instead of duplicating accruals. Late and out-of-order events are admitted up to a bounded lateness: the consumer tracks a per-partition watermark equal to the highest source_ts seen minus an allowed-lateness window, admits any event whose source_ts is at or above the watermark, and routes anything older to the late-event lane rather than dropping it. Replay is a first-class operation — because the log is durable and offsets are committed, an operator can reset a consumer group to an earlier offset to reprocess a corrupted window, and idempotency guarantees the replay is a no-op wherever facts already match. Schema evolution is governed by a schema registry: producers register each schema_version, only backward-compatible changes (new nullable fields, widened enums) deploy without a consumer bump, and breaking changes ship behind a dual-read window so in-flight events remain decodable. High-volume streams are consumed by a partitioned worker pool, which is why the consumer holds no cross-key state and shares its orchestration model with the async batch processing lane. The end-to-end mechanics of wiring a POS source to a live consumer are worked in streaming POS data with change data capture.

Drift Detection and Validation

A live stream is only trustworthy if it is continuously observed, because unlike a batch file its failures are silent — a stalled connector produces no error, just an absence. Drift detection runs control checks over the log and the consumer and quarantines anything anomalous rather than discarding it, so nothing is lost and nothing is silently accrued.

Catching these before facts reach the accrual engine is what preserves margin integrity: an offset gap caught in-flight is a targeted replay, while the same missing events discovered after settlement is a restated accrual and an audit finding. When any signal exceeds tolerance the consumer raises an exception carrying the offending event_id, partition, and offset so operations can reproduce the exact window.

Fallback and Dispute Routing

Poison events are inevitable on a live stream, so the consumer degrades rather than halts — a single unparseable event must never stall an entire partition. Failures are categorized deterministically and routed, never dropped. Structural failures (schema-invalid payloads, undecodable amounts, offsets that regress below the committed watermark for a key) route to a dead-letter topic with the raw event and a deterministic error code preserved for adjudication. Late-arrival failures (events beyond the allowed-lateness window) route to a separate late-event lane that reconciles them out-of-band rather than corrupting the watermark. Business-context failures (a change referencing an unregistered source table or a key with no agreement) are flagged for trade finance review and connect to the fallback routing logic that governs missing agreement terms. Each dead-lettered event carries its event_id, source offset, and a suggested remediation, and the dead-letter topic is retained so a fixed consumer can replay it once the defect is resolved. The escalation and adjudication mechanics — manual review, default-rate policy, dispute-to-deduction linkage into the settlement / deduction / financial close layer — are owned downstream; the consumer’s job is to name the correct lane, commit an audit entry, and keep the healthy partitions flowing.

Security and Access Boundaries

A CDC stream carries the source system’s most sensitive rows — negotiated pricing, party identifiers, transaction-level sales — continuously and in near-real time, so it is governed as a financial control, not a plumbing convenience. Access to the event log is enforced with topic ACLs: the connector holds write-only rights to the change topics, each consumer group holds read-only rights scoped to the topics it needs, and the dead-letter topic is separately restricted so quarantined payloads cannot leak. Data is encrypted in transit and at rest across the log, and field-level encryption protects monetary elements and party identifiers within events, so a broker operator cannot read cleartext amounts. Role-based access control tags travel with every canonical fact so authorization is enforced per field: ETL developers can deploy consumer versions and reset offsets in non-production but cannot mutate a committed fact, vendor managers can review dead-lettered events but cannot edit amounts, and trade finance analysts can adjudicate exceptions and export audit trails but cannot alter the log. Credential rotation is scheduled and automated — connector database credentials, broker SASL secrets, and consumer keys rotate on a fixed cadence, and the connector’s source-database account is granted only the replication privilege it needs, nothing broader. Immutable, hash-chained audit logs record every commit, quarantine, offset reset, and replay, anchored by source_offset and record_hash, so any tampering or gap surfaces as a chain break. Treating streaming and CDC ingestion as a versioned, access-governed control is what turns a live firehose of source changes into audit-ready financial data: lower settlement latency, earlier dispute detection, and a defensible trail from a committed row change to a posted accrual.

Frequently Asked Questions

How does the stream achieve exactly-once accrual on top of an at-least-once log? It does not rely on the broker for exactly-once; it makes the write idempotent. Every event yields a deterministic record_hash over its canonical business fields, and persistence is an upsert keyed on that hash. A redelivered event after a consumer restart or rebalance resolves to the same row and overwrites it identically, so duplicates and replays are no-ops. The system stays correct even if every event is delivered twice.

Why must monetary amounts travel as strings in change events? A JSON number is an IEEE-754 double, so encoding an amount as a bare number and decoding it in the consumer reintroduces binary-float rounding drift — the exact error batch pipelines avoid. Amounts are therefore serialized as strings and parsed straight into decimal.Decimal, keeping figures reproducible to the cent as IFRS 15 / ASC 606 accruals require, and any amount that arrives as a float is rejected as a broken producer.

What happens to a late or out-of-order event — is it dropped? Never dropped. The consumer tracks a per-partition watermark equal to the highest source commit time minus an allowed-lateness window. Events at or above the watermark are admitted and applied in per-key order; events older than the window are routed to a late-event lane and reconciled out-of-band. Ordering regressions within a key are refused by comparing the event’s source offset against the last committed offset for that key.

How does replay work without double-counting? Because the log is durable and offsets are committed, an operator can reset a consumer group to an earlier offset and reprocess a window. Idempotency makes this safe: every reprocessed event recomputes the same record_hash and upserts the same row, so replay is a no-op wherever facts already match and a clean correction wherever they were wrong. Replaying a full day reproduces the identical settlement figure.

Related

• Async Batch Processing — the file-based lane this streaming path complements, sharing the same canonical fact schema and worker-pool execution model.
• CSV & EDI Parsing Workflows — turns discrete vendor drops into typed records where CDC turns a mutating source table into a change log.
• Field Mapping Strategies — aligns the raw source column names inside change events to the canonical ledger schema.
• POS & ERP Sync Patterns — how streamed and batched records are pushed to and reconciled against downstream financial systems.
• Streaming POS Data with Change Data Capture — wiring a live POS source through a CDC connector to an idempotent consumer end to end.

Up one level: Data Ingestion & Normalization Pipelines