Data Cleansing and Transformation

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

Lesson Overview

A reseller sent TechLearningPro a nightly roster of team-pack enrollments. The file looked complete: one row per seat assignment, a price, a country, an email, and a timestamp. Finance asked why yesterday's revenue in fact_order_line was 3 percent below the payment provider. The answer sat in the file, not in the warehouse engine. One price arrived as 49.90, another as "$49.90", and a third as the float 49.89999999999999 after a partner system stored money in binary floating point. Twelve rows repeated the same order_line_id because the partner retried the export. Country values were US, USA, United States, and a blank. Two learners changed country inside the same file, and the cleanse job kept only the last row, so the type-2 dimension from Dimensional Modeling and Data Warehousing never saw the earlier version.

Data Quality Engineering tells you the file is unfit to publish. It does not, by itself, produce the typed, deduplicated, grain-correct rows the warehouse load expects. Cleansing and transformation are that production step. Cleansing repairs representation: types, codes, whitespace, nulls, and duplicates, and it quarantines rows that cannot be repaired without inventing a business fact. Transformation reshapes what remains into the contract the next table already declared: integer cents, UTC timestamps, ISO country codes, one surviving row per fact grain, and a change record that a slowly changing dimension can apply without erasing history.

The load that follows has to be idempotent. A retry, a backfill, and a second delivery of the same file must leave analytics.fact_order_line with one row per order_line_id and the same cents. If the job is safe to run twice, orchestration in the next lessons can retry it. If it is not, every retry double-counts revenue.

This lesson stays on a single machine and in PostgreSQL. The rules are the same rules a Spark job will apply later, when the roster no longer fits one process. Module 3, beginning with Big Data Fundamentals, distributes the work. It should not renegotiate what a cent is, which duplicate wins, or which attribute is allowed to overwrite history.

Learning Objectives

By the end of this lesson, you should be able to:

1. Separate a quality check, a cleansing rule, and a structural transformation, and place each in the path from landing to curated.
2. Parse money with a decimal type and store integer cents, and explain what binary floating point does to a price.
3. Classify an empty field as unknown, not applicable, or a failed parse, and quarantine the third.
4. Standardize emails, country codes, and timestamps with rules you can replay.
5. Deduplicate on a declared key and a declared winner order, including a tie-break.
6. Shape a transform so a type-2 dimension still receives every version the source actually contained.
7. Distinguish an event-shaped source from a snapshot-shaped source, and say what history each one can support.
8. Load a fact with an idempotent merge on the grain key so a replay does not insert a second row.
9. Keep enrichment joins from multiplying a measure, and reject a join that changes grain.
10. Write a row hash that makes "no meaningful change" detectable on replay.
11. Route irreparable rows to a quarantine table with a reason code, a source locator, and no silent drop.
12. State which of these rules move unchanged into a Spark job, and which only the runtime changes.

Prerequisites

• Data Quality Engineering: validity, completeness, and the decision to stop a publish. This lesson is what you do to the rows after a check, and which repairs are honest.
• Dimensional Modeling and Data Warehousing: fact_order_line grain, surrogate keys, and type-2 country and title. The transform must serve that contract.
• Advanced SQL and Window Functions: ROW_NUMBER for a dedup window, and the habit of stating the ORDER BY that picks a winner.
• Building Batch Data Pipelines with Python: a batch job with a landing path, a run id, and a failure that does not publish a partial curated table.
• Change Data Capture Fundamentals: an event already has a sequence. A nightly file often does not, and the cleanse has to say so.
• Comfort with decimal.Decimal and with PostgreSQL INSERT ... ON CONFLICT or MERGE (PostgreSQL 15+).

Detailed Core Concepts

Cleansing, transformation, and checks are three different decisions

A check observes and returns pass or fail. " extended_amount_cents is null on 40 rows" is a check. A cleansing rule changes representation without changing the business event: trim the email, map USA to US, parse 49.90 into 4990 cents. A transformation changes shape to match a downstream contract: drop the seat columns off an order-line grain, compute extended_amount_cents from quantity, unit price, and discount, or emit a dimension change row with effective_start.

Mixing them produces two failures. Teams "fix" a failed check by deleting the row inside the check, so the audit trail of the failure disappears. Or they put a grain change inside a function named clean_email, and reviewers cannot see that revenue moved. Name the stage for the decision it makes. In this course the stages are:

Quarantine is a first-class output of cleanse and conform, not an exception log you might lose. A quarantined row has the source locator (file, byte offset or row number), the run id, the reason code, and the original payload. A later repair reads that table. It does not reread a file that the partner has already replaced.

Types are part of the contract

The warehouse fact stores money as bigint cents. The landing file does not. The cleanse is the only place that conversion is allowed to happen, so every downstream job sees one type.

Binary floating point cannot represent most decimal fractions. 0.10 + 0.20 is not 0.30 in IEEE-754 double. A price that round-tripped through a JSON number or a Python float can become 49.89999999999999. If you then multiply by 100 and cast to int toward zero, you store 4989 and lose a cent. Enough lines, and finance sees the 3 percent gap as a mysterious leak. The repair is to parse the original text with a decimal type, scale to cents with an explicit rounding mode, and store an integer. If the field already arrived as a JSON number, you have already lost the decimal; quarantine it or require the partner to send a string. Do not "fix" a float by rounding after the fact and calling the result exact.

Timestamps get the same treatment. A value with no offset is not UTC just because the server is in UTC. Record the assumption in the job: "partner file timestamps are America/New_York wall time" or "values with no offset are UTC." Convert to timestamptz in UTC before the row leaves staging. A date-only business date stays a date. Do not store "midnight UTC" and later treat it as the partner's local midnight; those are different instants for a US retailer.

Booleans in files arrive as Y, N, true, 1, and empty. Map the known set. Anything else is a failed parse, not false. Empty is unknown, and unknown is not false. A false order_paid_flag that was actually blank will drop the line from revenue if the reconciliation filter is "paid only."

Nulls have a reason

Three empties look identical in a CSV and mean different things.

• Unknown. The partner did not send country. You may load a dimension attribute as null and let the report say "unknown." You may not invent US because most learners are in the US.
• Not applicable. A self-serve line has no reseller id. Null is correct, and a check that demands a reseller id on every row is the wrong check. Put the rule on the rows where the column applies.
• Failed parse. The field contained N/A, 49.90 USD, or a date the format string rejected. That is not a null you discovered. It is a value you refused. Quarantine the row with reason UNPARSEABLE_MONEY or UNPARSEABLE_TIMESTAMP. Replacing it with null hides a partner defect inside a sum that is now short.

Write the reason code down. NULL in the curated table is allowed only for unknown and not applicable, and the column description says which.

Standardization is a function, not a one-off script

Emails: trim, lowercase, and reject a string with no @ if the column is required. Do not "fix" jane@techlearningpro by appending .com. That is a guess. Country: map aliases (USA, United States, us) to ISO 3166-1 alpha-2, and quarantine the rest. Keep the map in a table, ref.country_alias, so a new alias is a data change. Phone numbers, if you store them, go to E.164 or stay out of the warehouse until you have a real parser. Free-text cleanup beyond trim and case is where well-meaning jobs corrupt names (McDonald to Mcdonald). Prefer a documented allow-list of transforms over a chain of regular expressions nobody can replay.

The function must be pure. Same input string, same output, no read of "today," no read of a mutable global. Enrichment that needs a reference table takes that table as an argument, pinned to a version or a loaded-at timestamp, so a replay next month uses the map that was current for that run or the map you explicitly choose. Write that choice in the run record.

Deduplication needs a key and a winner

A duplicate is two rows that claim the same grain. For fact_order_line the grain key is order_line_id. Twelve deliveries of the same line are one line. The winner is not "whichever row the database returns first." Declare the order: latest source_event_at, then highest ingest_sequence as a tie-break. ROW_NUMBER() OVER (PARTITION BY order_line_id ORDER BY source_event_at DESC, ingest_sequence DESC) with rn = 1 is that declaration. SELECT DISTINCT without an order is not, because it does not say which price survives when the retries disagree.

If two rows share the key and disagree on extended_amount_cents, you still need a policy. Latest-event-wins is reasonable for a webhook retry that corrected a price. It is dangerous if the "later" row is a partial retry with a zero price. When the cents differ, quarantine the key and let a person or a stricter rule decide. Silent latest-wins on money is how a zero-price retry erases a sale.

Duplicates at the wrong grain are a different bug. Twelve seats are not twelve copies of the line. Deduplicating seats by order_line_id deletes assignments. Deduplicating lines by email collapses two purchases by the same learner. The key is the grain key from the dimensional model, not a column that happens to repeat.

SCD-friendly transforms keep versions

A type-2 attribute changes by closing one dimension row and inserting another. The cleanse supports that only if it still has the versions. Two shapes of source:

Event or change feed. Each row is a change with learner_id, the new country, and observed_at. The cleanse standardizes country and emits every change, ordered by observed_at. It does not collapse to the learner's last country. The dimension loader from lesson 10 walks those changes and writes surrogate versions. Collapsing to "current" inside the file throws away a country that was true for a sale earlier the same day.

Snapshot file. Each row is the learner as of file time, with no history inside the file. You can diff the snapshot against the current dimension row and emit a change when country or email differs. You cannot recover a change that flipped and flipped back between snapshots. Say that in the pipeline document. A daily snapshot supports daily type-2, not intra-day type-2. Pretending otherwise writes a dimension that looks historical and is not.

Type-1 attributes are allowed to overwrite. Email might be type 1 while country is type 2. The staging row can carry both, and the loader applies the policies attribute by attribute. The cleanse does not apply them by updating dim_learner in place. One writer owns the dimension.

Course title works the same way. A catalog file that contains today's title is a snapshot. Emitting only the title, and letting the loader compare it to the current version, is the SCD-friendly shape. Overwriting dim_course.title in the cleanse job skips the surrogate and restates history.

Idempotent loads

Idempotent means a second successful run with the same inputs leaves the target in the same state as the first. For an insert-only fact, the practical form is a merge on the grain key:

• Missing key: insert.
• Same key, same row hash: do nothing.
• Same key, different hash: update the measure columns if corrections are allowed, or quarantine if the fact is frozen after publish.

The hash covers the columns that define "the same business row": cents, quantity, dimension business keys, business date. It does not cover loaded_at or the warehouse surrogate, or every replay looks like a change.

The merge is one statement, or one transaction. A delete-all-then-insert for the business date is also idempotent if the delete and insert commit together and the job only deletes rows it is responsible for. It is a poor default on a large fact, and it is unsafe if two runs overlap. Prefer a key-based merge and a lock or a single active run per target, which the orchestrator will enforce later.

A load that inserts and then "checks the count" is not idempotent. The check passes the first time and the second run doubles the count. Retries in lesson 21 will run that job again. Design the merge now, before the scheduler exists.

Enrichment must preserve grain

Adding the course track name by joining dim_course is enrichment. The join key is the course business key, and the course side must be one row per key at the moment you mean (the current version, or the version effective at sale time). A join to every historical version multiplies the line and multiplies revenue. If you need the title as of the sale, join on business key plus an effective-date predicate that returns one version, or defer the join to query time and store only course_id on the staged fact. Storing the surrogate is the loader's job after the dimension load, not the cleanse job's guess.

Exchange rates, if you convert, are a reference table with a date. Pin the rate row in the run. A rate that changes tomorrow must not change yesterday's fact on replay unless finance asked for a restatement.

What moves to Spark unchanged

The pure functions, the winner order, the quarantine reason codes, the cents rule, and the merge key do not change when the data no longer fits one Python process. What changes is the engine: partitions, shuffle, and file layout, which Big Data Fundamentals and the Spark lessons take up. A Spark job that reinterprets order_line_id or sums seats into revenue is a new system, not a port. Keep the unit tests from lesson 23 on the functions so the port has something to match.

Architecture and Design Diagrams

Raw files land once. Checks can reject a file before cleanse, or cleanse can quarantine rows and still publish the good grain. The curated fact is written by one merge. The dimension path keeps change rows instead of a single current snapshot when the source is a feed.

mermaidCopy
flowchart LR
    subgraph landing [Landing]
        File[Partner file and manifest]
    end
    subgraph stage [Staging in PostgreSQL]
        Raw[raw.order_line_landing]
        Typed[staging.order_line_typed]
        Changes[staging.learner_change]
        Quar[quarantine.order_line]
    end
    subgraph curated [Curated warehouse]
        Dim[analytics.dim_learner]
        Fact[analytics.fact_order_line]
    end
    File --> Raw
    Raw --> Typed
    Raw --> Quar
    Typed --> Changes
    Typed --> Fact
    Changes --> Dim
    Dim --> Fact

The dimension load reads staging.learner_change in observed_at order. The fact load runs after dimensions for that run, and it resolves surrogates. A fact whose learner business key is unknown follows the lesson 10 rule: stub or quarantine, and do not invent a surrogate inside the cleanse function.

mermaidCopy
flowchart TD
    Src[Source row] --> Parse[Parse types]
    Parse -->|parse failed| Q1[Quarantine UNPARSEABLE]
    Parse --> Std[Standardize codes]
    Std -->|code unknown| Q2[Quarantine UNKNOWN_CODE]
    Std --> Grain{Grain key present?}
    Grain -->|no| Q3[Quarantine MISSING_KEY]
    Grain -->|yes| Dedup[Dedup by grain key and winner order]
    Dedup -->|money conflict| Q4[Quarantine CONFLICTING_MEASURE]
    Dedup --> Shape{Target}
    Shape -->|fact| Merge[Idempotent merge on order_line_id]
    Shape -->|type-2 attribute| Emit[Emit every change with observed_at]
    Emit --> Scd[Dimension loader closes and inserts versions]

A snapshot source replaces the "emit every change" branch with a diff against the current dimension row. Draw that branch in the design review when the file has no event time inside the day. Leaving it off the diagram is how intra-day history gets implied and then lost.

Technical Explanation

Parsing money and time

The landing column is text, even if the header says price. A cleanse function accepts the original string.

pythonCopy
from decimal import Decimal, InvalidOperation, ROUND_HALF_EVEN

class ParseError(Exception):
    def __init__(self, reason: str, raw: str):
        super().__init__(reason)
        self.reason = reason
        self.raw = raw

def money_to_cents(raw: str) -> int:
    """Parse a decimal string into integer cents. Half-even at the cent."""
    text = raw.strip().replace(",", "")
    if text.startswith("
quot;):
        text = text[1:]
    try:
        amount = Decimal(text)
    except InvalidOperation as exc:
        raise ParseError("UNPARSEABLE_MONEY", raw) from exc
    cents = (amount * Decimal(100)).quantize(Decimal("1"), rounding=ROUND_HALF_EVEN)
    return int(cents)

ROUND_HALF_EVEN is a choice. Banker's rounding is stable and documented. Half-up is also a choice. Pick one, put it in the function, and test 1.005 and 1.015 if you ever scale more than two places. Do not call float.

For a value that is already a Python float because a JSON parser made it one, refuse it:

pythonCopy
def require_money_text(value: object) -> str:
    if isinstance(value, float):
        raise ParseError("MONEY_WAS_FLOAT", repr(value))
    if not isinstance(value, str):
        raise ParseError("MONEY_NOT_TEXT", repr(value))
    return value

Timestamps: datetime.fromisoformat on Python 3.11 accepts offsets. If the result is naive, attach the partner zone you wrote down and convert to UTC. Store both the UTC instant and the business date the partner meant, when those differ.

Standardization tables

sqlCopy
CREATE TABLE ref.country_alias (
    alias          text PRIMARY KEY,
    country_code   char(2) NOT NULL
);

INSERT INTO ref.country_alias (alias, country_code) VALUES
    ('us', 'US'),
    ('usa', 'US'),
    ('united states', 'US'),
    ('in', 'IN'),
    ('india', 'IN');

The cleanse lowercases and trims, then looks up. A miss is UNKNOWN_COUNTRY, not a pass-through of the raw string into dim_learner.country_code, where it will fail every ISO join later. Email standardization stays in code because the rule is smaller than a table: trim, lowercase, and a single @.

Dedup and conflict detection in SQL

PostgreSQL DISTINCT ON keeps the first row in the ORDER BY. The order is the winner rule.

sqlCopy
CREATE TABLE staging.order_line_typed (
    order_line_id          text NOT NULL,
    order_id               text NOT NULL,
    course_id              text NOT NULL,
    learner_id             text,
    business_date          date NOT NULL,
    quantity               integer NOT NULL CHECK (quantity > 0),
    unit_price_cents       bigint NOT NULL,
    discount_cents         bigint NOT NULL DEFAULT 0,
    extended_amount_cents  bigint NOT NULL,
    source_event_at        timestamptz NOT NULL,
    ingest_sequence        bigint NOT NULL,
    row_hash               text NOT NULL,
    run_id                 uuid NOT NULL,
    CONSTRAINT extended_matches CHECK (
        extended_amount_cents = quantity * unit_price_cents - discount_cents
    )
);

Before this table is filled, a conflict query finds keys whose surviving cents are ambiguous:

sqlCopy
WITH ranked AS (
    SELECT
        order_line_id,
        extended_amount_cents,
        ROW_NUMBER() OVER (
            PARTITION BY order_line_id
            ORDER BY source_event_at DESC, ingest_sequence DESC
        ) AS rn,
        COUNT(*) OVER (PARTITION BY order_line_id) AS copies,
        COUNT(DISTINCT extended_amount_cents) OVER (
            PARTITION BY order_line_id
        ) AS distinct_amounts
    FROM staging.order_line_parsed
)
INSERT INTO quarantine.order_line (order_line_id, reason_code, run_id, payload)
SELECT order_line_id, 'CONFLICTING_MEASURE', :run_id, NULL
FROM ranked
WHERE copies > 1
  AND distinct_amounts > 1;

Keys inserted there are excluded from the winner select. Keys with copies that agree on cents keep rn = 1. Agreement means retries are safe. Disagreement means a human looks at the payload you also stored from landing.

Row hash

Hash the business columns in a fixed order with a fixed encoding. md5 is enough to detect change. It is not a security control.

sqlCopy
md5(concat_ws('|',
    order_line_id,
    order_id,
    course_id,
    coalesce(learner_id, ''),
    business_date::text,
    quantity::text,
    unit_price_cents::text,
    discount_cents::text,
    extended_amount_cents::text
)) AS row_hash

Include the separator so 12|3 and 1|23 cannot collide across fields. concat_ws skips nulls in a way that can hide a null if you are not careful: coalesce nullable fields to a token such as <null> if null versus empty must hash differently. For learner_id, a null learner on a team line is meaningful. Use a token.

Idempotent fact merge

sqlCopy
INSERT INTO analytics.fact_order_line (
    order_line_id, order_id, business_date, course_id, learner_id,
    quantity, unit_price_cents, discount_cents, extended_amount_cents,
    row_hash, loaded_at, run_id
)
SELECT
    order_line_id, order_id, business_date, course_id, learner_id,
    quantity, unit_price_cents, discount_cents, extended_amount_cents,
    row_hash, now(), run_id
FROM staging.order_line_winners
ON CONFLICT (order_line_id) DO UPDATE
SET
    quantity = EXCLUDED.quantity,
    unit_price_cents = EXCLUDED.unit_price_cents,
    discount_cents = EXCLUDED.discount_cents,
    extended_amount_cents = EXCLUDED.extended_amount_cents,
    row_hash = EXCLUDED.row_hash,
    loaded_at = now(),
    run_id = EXCLUDED.run_id
WHERE analytics.fact_order_line.row_hash IS DISTINCT FROM EXCLUDED.row_hash;

The WHERE on the update avoids a write when the replay is identical, which keeps loaded_at stable and keeps vacuum quieter. A second run inserts nothing and updates nothing. That is the property you will test in lesson 23.

If finance freezes a closed month, add AND analytics.fact_order_line.business_date > :open_period_end to the update, and quarantine hash changes inside the freeze instead of applying them. A refund after close is a new fact, as lesson 10 already required. This merge must not become the back door that edits September.

SCD change records from a feed

sqlCopy
CREATE TABLE staging.learner_change (
    learner_id         text NOT NULL,
    email_normalized   text,
    country_code       char(2),
    observed_at        timestamptz NOT NULL,
    run_id             uuid NOT NULL,
    PRIMARY KEY (learner_id, observed_at)
);

One row per observed change, not per learner. The dimension loader reads ORDER BY learner_id, observed_at and applies type 2 for country_code and type 1 for email if that is the written policy. The primary key makes a replay of the same observation a no-op via ON CONFLICT DO NOTHING.

From a snapshot, the insert is a diff:

sqlCopy
INSERT INTO staging.learner_change (learner_id, email_normalized, country_code, observed_at, run_id)
SELECT s.learner_id, s.email_normalized, s.country_code, s.snapshot_at, :run_id
FROM staging.learner_snapshot AS s
LEFT JOIN analytics.dim_learner AS d
  ON d.learner_id = s.learner_id
 AND d.is_current
WHERE d.learner_key IS NULL
   OR d.country_code IS DISTINCT FROM s.country_code
   OR d.email_normalized IS DISTINCT FROM s.email_normalized;

IS DISTINCT FROM treats null as a value, so a newly missing email is a change. The snapshot timestamp is the effective start of the new version. Document that changes between snapshots are invisible.

Quarantine table

sqlCopy
CREATE TABLE quarantine.order_line (
    quarantine_id  bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    run_id         uuid NOT NULL,
    source_file    text NOT NULL,
    source_row     integer NOT NULL,
    order_line_id  text,
    reason_code    text NOT NULL,
    raw_payload    jsonb NOT NULL,
    created_at     timestamptz NOT NULL DEFAULT now()
);

Reason codes used in this lesson: UNPARSEABLE_MONEY, MONEY_WAS_FLOAT, UNPARSEABLE_TIMESTAMP, UNKNOWN_COUNTRY, MISSING_KEY, CONFLICTING_MEASURE, GRAIN_MULTIPLIED, CLOSED_PERIOD_CORRECTION. A dashboard on counts by reason_code and day is the operational view. A quarantine row is not a failure of the whole file unless the contract says the file is atomic. TechLearningPro's reseller file is row-atomic: good lines publish, bad lines wait. A payments file that must balance to the cent is file-atomic: any quarantine fails the run and publishes nothing. Choose per source and write it on the pipeline.

Reconciliation after the merge

The cleanse is not done when the merge returns. Sum extended_amount_cents on the winners for the business date and compare to the typed staging sum, which you already checked against quantity * unit_price_cents - discount_cents. Then compare the warehouse sum for that date to the provider settlement at the same grain, using the payment filter from lesson 10. A mismatch fails the run after the merge only if you can roll back. Prefer running the merge and the reconciliation in one transaction for that date partition, or write the fact to a staging table and swap when the sums match. Publishing first and alerting later is how the 3 percent gap reaches the board pack.

Practical Implementation or Design Walkthrough

Problem. The reseller file enrollments_2026-03-01.csv must become idempotent fact_order_line rows and SCD-friendly learner changes for business date 2026-03-01.

Source sample. Header: order_line_id,order_id,course_id,learner_id,email,country,qty,unit_price,discount,event_time,row_num.

textCopy
L1,O9,C3,U7,Ava@Example.com,USA,1,49.90,0,2026-03-01T15:04:00-05:00,1
L1,O9,C3,U7,ava@example.com,US,1,49.90,0,2026-03-01T15:06:00-05:00,2
L2,O9,C4,U8,Ben@Example.com,United States,12,$588.00,0,2026-03-01T16:00:00-05:00,3
L3,O10,C3,,team@example.com,,1,49.9,0,2026-03-01T16:10:00-05:00,4
L4,O11,C5,U9,nope,Atlantis,1,10.00,0,not-a-time,5

Target. L1 once, at 4990 cents, country US, email lowercased. The earlier and later L1 rows agree on money, so the later event wins and the duplicate is not a conflict. L2 is a team line: 12 × 4900 is wrong because the file already sent an extended unit of $588.00 for the pack. The column is named unit_price and the value is the pack. That is a contract bug. The walkthrough treats unit_price as the line's unit price only when qty is the multiplier the dimensional model defined. For a team pack, lesson 10 says quantity can be 12 seats or 1 pack. You must not guess. Here the file's own header is ambiguous, so L2 is quarantined as AMBIGUOUS_PACK_PRICE until the partner confirms whether 588.00 is the pack or the seat. Inventing 58800 * 12 is the twelve-times bug again. L3 has a null learner, which is legal for a team-owned line if learner_id is optional; country blank becomes unknown, not a quarantine, if the column is optional. L4 is UNPARSEABLE_TIMESTAMP and UNKNOWN_COUNTRY.

Implementation sketch. A Python function processes one dict and returns either a typed record or a quarantine record. It does not connect to a database. The database load is a separate step so tests can call the function with fixtures.

pythonCopy
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal, InvalidOperation, ROUND_HALF_EVEN

COUNTRY_ALIASES = {
    "us": "US",
    "usa": "US",
    "united states": "US",
    "in": "IN",
    "india": "IN",
}

@dataclass(frozen=True)
class TypedLine:
    order_line_id: str
    order_id: str
    course_id: str
    learner_id: str | None
    email_normalized: str | None
    country_code: str | None
    quantity: int
    unit_price_cents: int
    discount_cents: int
    extended_amount_cents: int
    source_event_at: datetime

@dataclass(frozen=True)
class Quarantine:
    reason_code: str
    source_row: int
    detail: str

def _cents(raw: str) -> int:
    text = raw.strip().replace("
quot;, "").replace(",", "")
    amount = Decimal(text)
    return int((amount * 100).quantize(Decimal("1"), rounding=ROUND_HALF_EVEN))

def parse_line(row: dict, source_row: int) -> TypedLine | Quarantine:
    if not row.get("order_line_id"):
        return Quarantine("MISSING_KEY", source_row, "order_line_id")
    try:
        event_at = datetime.fromisoformat(row["event_time"])
    except (TypeError, ValueError):
        return Quarantine("UNPARSEABLE_TIMESTAMP", source_row, row.get("event_time", ""))
    if event_at.tzinfo is None:
        return Quarantine("NAIVE_TIMESTAMP", source_row, row["event_time"])
    country_raw = (row.get("country") or "").strip().lower()
    country = COUNTRY_ALIASES.get(country_raw) if country_raw else None
    if country_raw and country is None:
        return Quarantine("UNKNOWN_COUNTRY", source_row, row["country"])
    try:
        qty = int(row["qty"])
        unit = _cents(row["unit_price"])
        discount = _cents(row["discount"] or "0")
    except (InvalidOperation, ValueError, KeyError) as exc:
        return Quarantine("UNPARSEABLE_MONEY", source_row, str(exc))
    if qty == 12 and unit % 12 == 0 and unit > 10000:
        # Ambiguous pack: a large "unit" with seat quantity. Do not multiply.
        return Quarantine("AMBIGUOUS_PACK_PRICE", source_row, row["unit_price"])
    email = (row.get("email") or "").strip().lower() or None
    return TypedLine(
        order_line_id=row["order_line_id"],
        order_id=row["order_id"],
        course_id=row["course_id"],
        learner_id=(row.get("learner_id") or "").strip() or None,
        email_normalized=email,
        country_code=country,
        quantity=qty,
        unit_price_cents=unit,
        discount_cents=discount,
        extended_amount_cents=qty * unit - discount,
        source_event_at=event_at,
    )

The pack heuristic is deliberately conservative and local to this file's known defect. A better contract is a column price_kind of unit or extended. Until that column exists, quarantine beats a silent multiply. After parse, group by order_line_id, apply the winner order, and emit conflicts when cents differ. Load winners with the ON CONFLICT statement. Replay the function on the same rows and show the typed output is identical; replay the merge and show one database row.

Operational notes. Record run_id, source file checksum, and row counts: landed, typed, quarantined, inserted, updated, unchanged. Alert when CONFLICTING_MEASURE or AMBIGUOUS_PACK_PRICE is nonzero. Keep the landing file. The quarantine payload is your ability to repair without asking the partner to resend.

Enterprise Real-World Scenarios

Reseller roster and the payment-provider gap

TechLearningPro's reseller sends 50,000 to 200,000 assignment rows a night, one file, expected by 05:00 UTC so finance can see the prior business date at 07:00 UTC. The file is row-atomic for personal-data defects and file-atomic for money: if quarantined lines exceed 0.5 percent of declared file revenue, the run publishes nothing and pages the partner manager. Country and email are standardized. Money is parsed from text. order_line_id dedup keeps the latest event when cents agree. The fact merge is idempotent, so the 06:10 retry after a lock timeout does not double revenue. Reconciliation compares warehouse cents for the business date to the provider's settled cents for the same order ids. The 3 percent gap in the opening story was float cents plus a pack multiplied by twelve. Both are gone when the parse refuses floats and the pack row is quarantined until the contract says what unit_price means. Security: the landing bucket is readable by the ingest role only; emails in quarantine are still personal data, so the quarantine table inherits the same access policy as dim_learner, not a world-readable log.

Learner profile feed and country history

The product database emits a nightly change feed of learner profile updates, about 20,000 changes on a busy day, each with learner_id, country, email, and updated_at. Marketing asks for revenue by country of the learner at the time of purchase. That is a type-2 question. The cleanse maps country, lowercases email, and writes every change to staging.learner_change. It does not reduce the feed to one row per learner before the dimension load. A learner who moved from IN to US on the same date as a purchase produces two change rows; the fact for that purchase still points at the surrogate that was current at source_event_at, which the dimension loader computes from effective dates. A second, "simpler" job that updates country_code in place on dim_learner would restate last year's revenue into the new country. The feed is the source of history. The snapshot diff pattern is reserved for the legacy CSV that has no updated_at inside the day, and the catalog description of that dimension says "versioned at daily snapshot grain."

Course catalog title corrections

A content operator fixes a course title twice in one afternoon. The catalog CDC feed contains both titles with timestamps. The conform step emits two course_change rows. The type-2 loader creates two versions. A cleanse that keeps the max timestamp only would still be right for "current title," and it would be wrong for a report that must show the title as sold at 13:00. Because lesson 10 marked title as type 2, the transform keeps both. Volume is tiny. The point is the rule, not the cluster size. The same function, with the same tests, is what you will later run as a Spark map when a high-volume event stream replaces the file.

Best Practices

• Land bytes unchanged, with a checksum, before any parse.
• Parse money from text with Decimal and store cents. Refuse floats.
• Make every standardization rule a pure function plus a versioned reference table.
• Declare the dedup key, the winner order, and the tie-break in SQL that a reviewer can read.
• Quarantine measure conflicts. Do not let latest-wins rewrite cents when the values disagree.
• Emit one SCD change per source observation for type-2 attributes. Collapse only attributes you have classified as type 1, and collapse them in the dimension loader.
• Diff snapshots against the current dimension explicitly, and document the history you cannot see.
• Merge facts on the grain key. Include a row hash so identical replays do not update loaded_at.
• Keep enrichment from changing grain. One dimension version per fact row.
• Reconcile sums in the same transaction as the publish, or publish by swap after the sums match.
• Store reason codes, source locators, and raw payloads in quarantine. Restrict that table like the personal data it contains.
• Fix the partner contract when a heuristic appears. A heuristic is a temporary quarantine, not a permanent formula.
• Write the rules so a later Spark job can call the same decisions per row. Keep I/O and shuffle outside those functions.

Common Mistakes and Anti-Patterns

Casting prices through float. The file looks numeric, so the job calls float(price) * 100. Consequence: cent-level drift and a reconciliation gap that no retry will fix, because the error is stable. Fix: Decimal on the original text, and reject JSON numbers for money fields.

SELECT DISTINCT as a dedup strategy. The database drops duplicate tuples and keeps an arbitrary survivor when non-key columns differ. Consequence: a zero-price retry can win. Fix: ROW_NUMBER with an explicit order, and a conflict quarantine when measures disagree.

Keeping the last row per learner before a type-2 load. The file contained the history. Consequence: the dimension has one version per day at best, and intra-file moves disappear. Fix: emit every change with observed_at.

Filling unknown country with the mode of the file. Nulls look untidy in a demo. Consequence: revenue is assigned to a country the learner did not claim. Fix: store unknown, and split "unknown" in the report.

Deleting quarantined rows from landing. The curated table looks clean. Consequence: you cannot explain a short sum or replay a repair. Fix: quarantine table plus immutable landing.

Joining all SCD versions during enrichment. The join is "just to get the title." Consequence: the line duplicates once per historical title and the sum explodes. Fix: store the business key and resolve one surrogate in the loader, or constrain the join to a single effective row.

Delete-and-reload without a transaction or a partition lock. Two retries overlap. Consequence: lost rows or duplicates. Fix: key merge, one active run, and a transaction around the date you own.

Applying closed-period corrections in the same merge as open periods. A late file "fixes" September. Consequence: the board pack moves. Fix: quarantine CLOSED_PERIOD_CORRECTION and post refunds as their own fact.

Multiplying a pack price by seat quantity because both numbers are present. Consequence: the lesson 9 and lesson 10 bug returns inside the cleanse. Fix: one grain sentence on the staging table, and a quarantine when the file's price column is ambiguous.

Hidden current-time in the transform. datetime.now() becomes the event time when the parse fails. Consequence: replays and backfills stamp today's clock and scramble SCD order. Fix: event time from the row, or quarantine. Run id timestamps belong on the run record, not on the business event.

Troubleshooting and Design Review

Design review before a new source goes live:

1. What is the grain sentence of the staged row, and which source columns are illegal at that grain?
2. Which fields are text-parsed, and which rounding mode applies to money?
3. What is the dedup key, the winner order, and the conflict policy?
4. Which attributes are type 2, and does this source emit events or snapshots?
5. What is the merge key, and what does a second run change?
6. Which reason codes fail the file, and which only quarantine a row?
7. What reconciliation must match before publish?
8. Where will these functions run when the file grows past one machine?

Interview Questions and Answers

Q1. How is a cleansing rule different from a data-quality check?

A check reports whether rows meet a rule and should leave the row available for repair. A cleansing rule changes representation: type, case, code, trim. A transformation changes shape to the target grain. Putting a delete inside the check removes the evidence. Putting a grain change inside a function called clean_email hides a business decision. TechLearningPro lands the file, checks it, cleanses representation, conforms grain, then merges. Each stage has a different permission to change data.

Q2. Why store money as integer cents instead of double precision?

Binary floating point cannot represent most decimal currency fractions. A value that passes through float can drift by a fraction of a cent, and a cast to integer truncates it. Summed across a day of orders, finance sees a gap against the payment provider. Parse the partner's original decimal text with decimal.Decimal, quantize to cents with a documented rounding mode, and store bigint. If the payload already arrived as a JSON number, the decimal information may already be gone; quarantine or change the contract rather than rounding a float and calling it exact.

Q3. When should an empty country become null, and when should the row be quarantined?

Empty means unknown if the field is optional. The curated column stays null and the report shows an unknown bucket. A value that is present and not in the alias map is a failed standardization, reason UNKNOWN_COUNTRY, because writing the raw string into a char(2) column or guessing US invents a fact. A required country on a source whose contract says country is mandatory fails the row. The difference is whether the source promised a value.

Q4. Two webhook deliveries share an order_line_id and disagree on price. What wins?

Nothing automatic. If you declared latest-source_event_at wins, a partial retry with a zero price can erase a sale. TechLearningPro dedupes automatically only when the measure cents are equal, and quarantines the key with CONFLICTING_MEASURE when they are not. The winner order is still required for the equal case, with ingest_sequence as a tie-break, so the choice is not left to the storage engine. DISTINCT without ORDER BY is not a policy.

Q5. What does "SCD-friendly" mean for a cleanse job?

The job preserves every version the source actually contains for attributes classified as type 2. A change feed becomes one learner_change row per observation, with a standardized country and an observed_at, and the dimension loader opens and closes versions. The cleanse does not reduce the feed to the last country per learner. A snapshot file can only emit a diff against the current dimension row, and the pipeline must say that intra-snapshot history is absent. Type-1 attributes may overwrite later, in the loader, on purpose.

Q6. How do you make the fact load idempotent?

Merge on the grain key order_line_id. Insert when the key is absent. Update measure columns only when the business row_hash differs. Do nothing when the hash matches, so a retry does not bump loaded_at or create a second row. A unique constraint backs the merge. Delete-and-reload of a business date can also be idempotent inside one transaction, and it is easier to get wrong under overlapping runs. The orchestrator will retry this statement. The statement has to tolerate that.

Q7. A join to dim_course to attach the title doubles revenue. What happened?

The join matched more than one dimension row, usually every type-2 version of the course, or a fan-out from seats. Enrichment is required to return one row per fact grain. Store the course business key on the staged fact and resolve a single surrogate after the dimension load, with an effective-date predicate if the title as of sale is required on the fact. A multiplied staging table is a grain bug, reason GRAIN_MULTIPLIED, and it should fail the run before the merge.

Q8. What is the difference between an event-shaped source and a snapshot-shaped source for history?

An event-shaped source records changes with timestamps. Kept intact, it can build intra-day type-2 versions. A snapshot-shaped source records the entity as of file time. Diffing it to the current dimension versions the attribute at snapshot cadence only. A change that reverts before the next snapshot never appears. Both are usable. They are not interchangeable, and the catalog text for the dimension should state the cadence.

Q9. Why is a row hash useful if you already have a primary key?

The key makes the merge target unique. The hash says whether the business columns changed. Without it, an ON CONFLICT DO UPDATE rewrites every replayed row, moves loaded_at, and can look like a correction. With it, identical retries are no-ops, and a real price correction is visible as a hash change you can still block for closed periods. The hash covers business columns only, in a fixed field order, with an explicit null token.

Q10. When would you fail the whole file instead of quarantining bad rows?

When the file is a balancing set. A payments extract that must match a declared total cannot publish the good rows and shrug at the rest; the published sum would be short with a green job. TechLearningPro treats the reseller assignment file as row-atomic up to a threshold of quarantined revenue, and file-atomic above that threshold. Personal-data defects quarantine the row. Money-contract defects can fail the run. The choice is written per source, not implied by the framework.

Q11. A backfill of last Tuesday re-parses the file and the totals move. Where do you look first?

Any use of the wall clock inside the transform, a reference table that changed since Tuesday, a float parse, or a merge that updates closed dates. The function should depend on the row and on a pinned alias-table version. Event time comes from the file. If the country alias map gained a new row, decide whether history should be restated and, if not, pin the map version on the run. Then compare quarantine counts to the original run record.

Q12. Which parts of this design should a later Spark job preserve?

The pure parse, the cents rule, the reason codes, the dedup key and winner order, the SCD decision to keep versions, and the grain of the output row. Spark changes how rows are partitioned and written. It is not a reason to redefine extended_amount_cents or to collapse type-2 changes. Lesson 23 puts those rules in tests that do not import a cluster. Module 3 should run the same rules across partitions and then apply an equivalent idempotent write for the target table format.

Hands-On Exercises

Exercise 1. Parse, quarantine, and dedup

Problem. Implement parse_line and a dedup step for the five-row sample in the walkthrough, plus a second delivery of L1 with the same cents and a later timestamp.

Tasks.

1. Parse the sample. List typed rows and quarantine rows with reason codes.
2. Explain why L2 must not become 12 * 58800 cents.
3. Dedup L1. Show the surviving timestamp and that the cents match.
4. Add a sixth row: same L1, later timestamp, unit price 0. Show a CONFLICTING_MEASURE and no survivor.
5. Run the parse twice on the first sample and show identical typed output.

Acceptance. L1 survives once at 4990 cents until the zero-price conflict. L4 is quarantined. No float appears in the path.

Exercise 2. Idempotent merge and an SCD change list

Problem. Load winners into PostgreSQL and emit learner changes.

Tasks.

1. Create staging.order_line_typed, analytics.fact_order_line with a unique order_line_id, and staging.learner_change.
2. Merge the winners. Record inserted versus updated counts.
3. Run the merge again. Show zero inserts, zero updates, and a stable loaded_at when the hash matches.
4. Feed two country observations for one learner and show two change rows. State what the dimension loader will do with them. You do not need to re-implement the full type-2 loader if you describe the close-and-insert precisely.
5. Diff a snapshot that lacks observed_at history and write the limitation in one sentence on the table comment.

Rubric.

Lesson Summary

Cleansing makes rows truthful in their representation. Transformation makes them fit a grain the warehouse already declared. Neither step is allowed to invent a country, multiply a pack, or collapse type-2 history that the source actually sent. Money becomes integer cents through a decimal parse. Duplicates resolve through a documented winner order, and disagreeing money goes to quarantine with the raw payload. Facts merge on order_line_id so a retry is a no-op when the hash matches. Learner and course changes stay as a sequence the dimension loader can version.

Those properties are what make the next lesson possible. An orchestrator will run this job on a schedule, retry it, and backfill a business date. It can do that safely only because the cleanse and the merge are deterministic and idempotent. The same functions are the contract you will carry into Spark when a single Python process is no longer the right engine. The meaning of a cent does not change with the size of the file.

Further Reading

• Python, decimal — rounding modes and why currency code should not use binary floating point.
• PostgreSQL, INSERT ... ON CONFLICT and MERGE — the two idempotent write forms used in this lesson. MERGE needs PostgreSQL 15 or newer.
• PostgreSQL, SELECT DISTINCT ON — winner-take-one per key, which is only as good as the ORDER BY you write.
• ISO 3166-1 alpha-2 country codes, via the public code list you adopt into ref.country_alias. Standardize to the code, and keep aliases in your table.
• Kimball and Ross, The Data Warehouse Toolkit, slowly changing dimensions — the loader policy this transform must not undo.
• TechLearningPro, Dimensional Modeling and Data Warehousing and Data Quality Engineering — grain, type 2, and the checks that sit beside this cleanse.
• Next: Data Pipeline Orchestration Fundamentals, then Module 3 Big Data Fundamentals when these row rules need a distributed engine.