13. Data Ingestion Layer

The ingestion layer is where the framework’s most consequential work happens. In a data lake, ingestion is a transport concern: land the bytes, interpret later. In a Data Swimming Pool, ingestion performs canonicalization, entity resolution, and classification — and doing so once at entry, rather than repeatedly at query time, is the primary mechanism by which the correlation tax is eliminated.

This is the filtration system of the metaphor, and it is the difference between a maintained pool and a swamp.

13.1 Ingestion Workflow #

flowchart TB
    START(["Source event or batch"]) --> CONN["<b>1. Connect</b><br/>CDC · stream · file · API pull"]
    CONN --> RAW[("<b>2. Land raw</b><br/>raw.* topic + Bronze zone<br/><i>immutable, source-fidelity</i>")]

    RAW --> CTR{"<b>3. Contract<br/>validation</b>"}
    CTR -->|"schema violation"| Q1["Quarantine<br/><i>dlq.schema</i>"]
    CTR -->|"required field missing"| Q1
    CTR -->|"type mismatch"| Q1
    CTR -->|pass| QUAL{"<b>4. Quality<br/>assertion</b>"}

    QUAL -->|"range violation"| Q2["Quarantine<br/><i>dlq.quality</i>"]
    QUAL -->|"referential failure"| Q2
    QUAL -->|"freshness violation"| Q2
    QUAL -->|pass| DEDUP{"<b>5. Deduplicate</b>"}

    DEDUP -->|"duplicate business key"| DROP["Drop + count<br/><i>metric only</i>"]
    DEDUP -->|unique| CANON["<b>6. Canonicalize</b><br/>map to canonical event schema<br/>normalize units, currency, timezone<br/>assign event time + clock quality"]

    CANON --> ER["<b>7. Entity Resolution</b><br/>deterministic match → probabilistic match<br/>→ assign resolved entity ID<br/><i>retain original identifiers</i>"]

    ER --> ERCONF{"Resolution<br/>confidence?"}
    ERCONF -->|"high (≥0.95)"| CLASS
    ERCONF -->|"medium (0.70–0.95)"| CLASS2["Tag as<br/><i>provisional resolution</i>"]
    ERCONF -->|"low (<0.70)"| Q3["Quarantine<br/><i>dlq.resolution</i><br/>+ steward review queue"]
    CLASS2 --> CLASS

    CLASS["<b>8. Classify</b><br/>sensitivity tier · regulatory scope<br/>purpose tags · retention class"]
    CLASS --> POL{"<b>9. Ingestion<br/>policy check</b>"}
    POL -->|"prohibited source/purpose"| Q4["Reject + audit<br/><i>compliance event</i>"]
    POL -->|permit| LIN["<b>10. Emit lineage</b><br/>source → canonical event<br/>OpenLineage event"]

    LIN --> PUB[("<b>11. Publish</b><br/>canonical.* topic<br/>+ Silver zone")]
    PUB --> END(["Available for correlation"])

    Q1 & Q2 & Q3 --> NOTIFY["Notify source owner<br/>SLA-bound remediation"]
    NOTIFY -.reprocess after fix.-> CTR

    classDef ok fill:#183028,stroke:#4caf7d,color:#e6fff2
    classDef bad fill:#4a2020,stroke:#d07070,color:#ffeaea
    classDef proc fill:#0d4159,stroke:#22b3d6,color:#eaf9ff
    classDef store fill:#0a5570,stroke:#3fd0f0,stroke-width:2px,color:#ffffff
    class Q1,Q2,Q3,Q4,DROP bad
    class RAW,PUB store
    class CONN,CANON,ER,CLASS,LIN,CLASS2 proc
    class END ok

Figure 17. Data ingestion workflow. Eleven stages, four quarantine paths, and one drop path. The critical design property is that nothing reaches the correlation layer unvalidated, unresolved, or unclassified — the architectural expression of the filtration metaphor and the reason a Data Swimming Pool cannot degrade into a swamp through neglect alone.

13.2 Ingestion Patterns by Source Type #

Table 19. Ingestion patterns by source type. Two rows warrant particular attention: IoT sources require an explicit clock quality indicator because device clock skew routinely exceeds correlation window widths, and derived events require derivation-order tagging to prevent the feedback amplification described in Chapter 10.

13.3 Data Contracts #

A data contract is a machine-enforceable agreement between a source owner and the platform. The framework treats contracts as mandatory, not advisory.

Each contract declares:

The rationale is specific to correlation. In a conventional lake, a malformed record degrades one dataset. In a correlation architecture, a malformed record propagates: it may be incorrectly entity-resolved, forming false relationships with events belonging to a different entity, which then feed insights and potentially actions. Correlation amplifies data quality defects rather than absorbing them. Contract enforcement is therefore load-bearing rather than hygienic.

13.4 Entity Resolution #

Entity resolution is the single most valuable and most difficult function in the ingestion layer.

13.4.1 The Problem #

The same real-world entity appears across systems with different identifiers and inconsistent attributes:

Without resolution, correlating a service ticket with a payment failure and a device fault is impossible: no join key exists.

13.4.2 Resolution Strategy #

The framework specifies a three-tier cascade, executed in order:

Tier 1 — Deterministic matching. Exact match on authoritative identifiers: tax registration number, legal entity identifier, verified email, national identifier where lawful. High precision, limited recall. Every match is recorded with the rule that produced it.

Tier 2 — Probabilistic matching. Fellegi–Sunter-style probabilistic linkage or a learned pairwise classifier over normalized attributes (name similarity, address similarity, temporal co-occurrence, structural graph features). Produces a match probability. Blocked by candidate-generation keys to avoid all-pairs comparison — the same combinatorial discipline Principle 3 imposes on correlation.

Tier 3 — Graph-based resolution. Transitive closure over high-confidence pairwise matches, with community detection to identify entity clusters. Critically, this tier must include cluster-splitting logic: transitive closure over probabilistic matches produces over-merged clusters (“everything connects to everything”) unless actively constrained by cluster cohesion thresholds.

13.4.3 Handling Resolution Uncertainty #

The framework’s handling:

• High confidence (≥ 0.95): resolve and proceed.
• Medium confidence (0.70–0.95): resolve but tag as provisional. Correlations formed on provisional resolutions carry reduced confidence and are excluded from autonomy levels 3 and above.
• Low confidence (< 0.70): quarantine to a data steward review queue. Human adjudication is required. This queue’s depth is an operational metric, and a growing queue indicates deteriorating source quality or model drift.
• All resolutions are reversible. A resolution decision is an event with lineage. If it is later found incorrect, W2 retrospective re-correlation (Chapter 12) recomputes affected correlations.

13.5 Classification and Policy Tagging #

Every canonical event is tagged at ingestion with:

Table 20. Classification tags applied at ingestion. Permitted purposes and special category are the two tags that make correlation-time policy enforcement (Principle 8) possible: the correlation engine can refuse to relate a special-category event to a commercially-purposed scope because the tag travels with the event.

Classification is performed by a combination of contract declaration (the source owner declares what the field contains), automated detection (pattern and model-based identification of sensitive content), and inheritance (derived events inherit the most restrictive classification of their inputs).

13.6 Quarantine and Remediation #

Quarantine is a first-class operational concept, not an error log.

flowchart LR
    subgraph DLQ["Quarantine Queues"]
        direction TB
        D1["dlq.schema<br/><i>contract violation</i>"]
        D2["dlq.quality<br/><i>assertion failure</i>"]
        D3["dlq.resolution<br/><i>ambiguous entity</i>"]
        D4["dlq.policy<br/><i>prohibited ingestion</i>"]
    end

    subgraph TRIAGE["Triage"]
        direction TB
        T1{"Systemic or<br/>isolated?"}
        T2["<b>Systemic</b><br/>→ page source owner<br/>→ consider pausing source"]
        T3["<b>Isolated</b><br/>→ steward queue<br/>→ batch remediation"]
        T1 --> T2 & T3
    end

    subgraph ACT["Remediation"]
        direction TB
        A1["Fix at source"]
        A2["Amend contract"]
        A3["Manual adjudication"]
        A4["Accept exclusion<br/>+ document"]
    end

    REPRO["Reprocess from raw.*<br/><i>original fidelity retained</i>"]
    IMPACT["<b>Correlation impact assessment</b><br/>Which scopes ran on incomplete data?<br/>Which correlations need revision?"]

    DLQ --> TRIAGE --> ACT --> REPRO --> IMPACT

    classDef dlq fill:#4a2020,stroke:#d07070,color:#ffeaea
    classDef triage fill:#3d3313,stroke:#d4a636,color:#fff8e6
    classDef act fill:#183028,stroke:#4caf7d,color:#e6fff2
    class DLQ,D1,D2,D3,D4 dlq
    class TRIAGE,T1,T2,T3 triage
    class ACT,A1,A2,A3,A4,REPRO,IMPACT act

Figure 18. Quarantine and remediation flow. The final stage is what distinguishes this from conventional dead-letter handling: after reprocessing, the system must assess which correlation scopes executed on incomplete data during the outage and revise affected correlations. Silent data gaps produce confidently wrong correlations, so gap accounting is mandatory.

The gap accounting requirement. If a source was quarantining for four hours, every correlation scope including that source produced results on incomplete evidence during that window. The framework requires that:

1. Scopes affected by a source outage are marked degraded for the affected window.
2. Correlations produced during a degraded window carry a degradation flag.
3. After remediation, affected windows are re-correlated via W2.
4. Consumers that acted on degraded correlations are notified.

This is operationally demanding and is one of the framework’s genuine cost drivers. It is also non-negotiable: a correlation system that does not know when it was operating blind will assert conclusions it cannot support.

13.7 Ingestion Observability #

Table 21. Ingestion observability metrics. Volume deviation detection is the most operationally valuable, because a source that silently stops producing is the failure mode most likely to cause confidently wrong correlations — the system does not know it is missing evidence unless it is watching for absence.

Next: 14. Correlation Engine →