data-mapping-framework

A data modeling framework that ingests data from multiple platforms and formats, maps every heterogeneous source onto a single canonical model, and applies business rules (computation + validation) — producing a clean, validated dataset plus a rejection report.

This is the pattern I've applied across years of data engineering: reconcile messy inputs from different systems into one well-defined model, then enforce the business logic in one testable place. Domain and data here are synthetic (insurance policies); the architecture is the point.

The problem it models

The same entity — an insurance policy — arrives from four systems, each with its own column names, casing, types, and file format. Downstream analytics needs one consistent, validated model.

 broker CSV        legacy JSON API      warehouse SQL       lakehouse Parquet
 PolicyNo,Ctry…    {id, holder_name…}   policy_id,premium   policy_id,premium
      │                  │                    │                   │
      └──────────────────┴─────── map ────────┴───────────────────┘
                                   │
                          ┌────────▼────────┐
                          │ canonical model │   policy_id, holder, country,
                          │  (one schema)   │   premium, currency, start_date,
                          └────────┬────────┘   term_months
                                   │
                    computation rules → premium_eur, annualized, policy_year
                    validation rules  → accept / reject with a reason
                                   │
                      ┌────────────┴────────────┐
                      ▼                          ▼
                 valid dataset            rejected + rejected_by

How it's structured

Module	Responsibility
model.py	The canonical schema — the single source of truth
readers.py	One reader per platform: CSV, JSON, SQL (DuckDB), Parquet
mapping.py + sources.py	Declarative FieldMappings from each source to canonical
transforms.py	Reusable, coercing column transforms (upper, to_float, parse_date…)
rules.py + engine.py	ComputationRule / ValidationRule primitives + the run engine
policy_rules.py	A concrete BusinessRules subclass — one class per integration

Adding a new source is one SourceMapping; adding a new rule is one ValidationRule / ComputationRule. The canonical model and the engine never change.

Quickstart

pip install -r requirements.txt
PYTHONPATH=src python -m mapper.app

You'll see all four sources ingested into the canonical model, the valid rows (with derived premium_eur, annualized_premium_eur, policy_year), and the rejected rows each tagged with every rule they violated ("; "-separated, in rule order) — an error report that only names the first violation forces one fix-resubmit cycle per rule; naming all of them lets the sender fix the whole row in one pass.

You'll also see any cross-source key conflicts: a policy_id that arrived from more than one platform. Per-row validation can't catch this (each copy is valid on its own), and a per-source view can't either — only comparing keys across the stacked sources shows the collision, so it can be resolved deliberately instead of by whichever row landed last (detect_key_conflicts; the count is in report["key_conflicts"]).

Run with Docker

A container image is published to the GitHub Container Registry on every push:

docker run --rm ghcr.io/renatoaragon/data-mapping-framework:latest

Extending it

# 1. A new platform — just declare how its columns map to canonical.
from mapper.mapping import FieldMapping, SourceMapping
from mapper.transforms import as_string, to_float, upper

PARTNER_API = SourceMapping(
    name="partner_api",
    field_mappings=(
        FieldMapping("PolicyRef", "policy_id", as_string),
        FieldMapping("Insured",   "holder",    as_string),
        FieldMapping("Nation",    "country",   upper),
        FieldMapping("Gross",     "premium",   to_float),
        # ...
    ),
)

# 2. A new rule — add it to the BusinessRules subclass.
from mapper.rules import ValidationRule
ValidationRule("holder_present", lambda df: df["holder"].astype("string").str.len() > 0)

Tests

pytest -q

Covered: each reader (CSV / JSON / Parquet), mapping heterogeneous sources onto the canonical schema, the transforms' coercion behavior, the computation and validation rules, and an end-to-end run across all four sources. CI runs the suite on every push.

Design principles

Tools are the most disposable part of data engineering. pandas, DuckDB and the file formats on either side are all replaceable; the same framework would map JDBC sources or Avro onto the same canonical model with the same rules. What lasts is the shape: heterogeneous inputs reconciled to one schema before any logic runs, mapping expressed as data rather than branching code, and business rules as testable units. So this repo is organized around the model and the rules, not the readers that happen to feed them.

This is the integration layer of a data platform: the step that turns messy, multi-system inputs into the single conformed model everything downstream relies on, the clean input a pipeline like spark-retail-etl or a modelling layer like dbt-duckdb-analytics assumes it already has. The principles it holds to:

• One canonical model, upstream of everything. Every source conforms to it before a single rule runs, so downstream code deals with one schema, never four.
• Mapping is data, not code. Reconciliation lives in declarative FieldMappings, so a new source is one declaration and the engine never changes.
• Coerce, then reject with a reason. Bad values become NA and are rejected by a named rule, with every violation reported, instead of crashing the run or failing on the first error.

Design notes

• One canonical model — every source conforms to it before any logic runs, so downstream code deals with a single schema.
• Declarative mappings — the reconciliation lives in data (FieldMappings), not scattered if/else.
• Coerce, don't crash — transforms turn bad values into NA; the validation layer rejects them with a named reason instead of blowing up the pipeline.
• Rules as first-class objects — computation and validation are testable units, organized as one BusinessRules class per integration.

License

MIT — see LICENSE.