Cross-channel analytics often begins with a deceptively simple request: combine campaign performance from two advertising platforms. The challenge is that the exports may not agree on column names, units, identifiers, or even the meaning of a conversion.
In my public Retail Media Analytics Pipeline, I recently added a small Python ingestion layer that maps fictional Google-like and Meta-like CSV exports into a single nine-column contract. The implementation is available here, with tests and a successful GitHub Actions run for the October 24, 2026 commit.
Important scope: these are entirely simulated files, not actual vendor exports or live API integrations. The field names and measurement assumptions are invented for an inspectable engineering exercise.
1. Define the destination before mapping the sources
The downstream PySpark Bronze-to-Silver job expects the following canonical columns:
event_date,platform,campaign_id,campaign_name,
impressions,clicks,spend,conversions,revenue
~~~
A consistent *shape* lets the next stage parse the CSV, but it does not magically make attribution definitions comparable. That distinction matters.
The adapter's explicit source map captures the mechanical differences:
| Canonical field | Google-like CSV | Meta-like CSV |
| --- | --- | --- |
| Date | date | day |
| Campaign ID | campaign_id | ad_campaign_id |
| Campaign name | campaign_name | ad_campaign_name |
| Impressions | impressions | views |
| Clicks | clicks | link_clicks |
| Spend | cost_micros | spend_usd |
| Conversions | conversions | purchases |
| Revenue | conversion_value_usd | purchase_value_usd |
In this **mock contract**, both spend and revenue are USD. The Google-like spend field is stored as integer micro-USD; the Meta-like spend field is already in USD. These are assumptions of the fixtures, not general promises about every platform export.
The code uses a **SOURCES** dictionary to select the right input column for each canonical field. Adding a new source should mean defining a new mapping *and* its semantic contract, rather than scattering column renames throughout analytics transformations.
## 2. Normalize money deliberately
A classic ingestion bug is treating micro-units as dollars. For the synthetic Google-like input, **62,005,000 micro-USD equals $62.005**, which this adapter rounds to **$62.01**.
The implementation uses Python's Decimal rather than binary floating-point arithmetic:
~~~python
from decimal import Decimal, ROUND_HALF_UP
amount = Decimal("62005000") / Decimal(1_000_000)
usd = amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
assert str(usd) == "62.01"
~~~
The real helper also rejects negative, non-finite, and nonnumeric amounts, and checks that the micro-unit input is an integer. The Meta-like example **7.995 USD** becomes **8.00 USD** using the same explicit rounding policy.
Why does the rounding rule deserve a test? Because rounding at the **row level** before aggregation can differ from aggregating higher-precision source values and rounding at the end. The current portfolio contract chooses two-decimal normalized rows. A production system should document whether that choice is acceptable for reconciliation and billing.
See the [Python Decimal documentation](https://docs.python.org/3/library/decimal.html) for quantization and rounding behavior.
## 3. Reject invalid rows instead of quietly guessing
The adapter's **adapt_row(source, row)** function validates one observation before emitting canonical values. Its rules include:
- A real ISO calendar date in YYYY-MM-DD form, not merely a string that looks date-like.
- Nonempty campaign ID and name.
- Nonnegative integer impressions, clicks, and conversions.
- Clicks no greater than impressions **for this simplified example**.
- Finite, nonnegative spend and revenue.
The date check combines a format expression with a calendar parser:
~~~python
from datetime import date
import re
day = row["date"].strip()
if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", day):
raise ValueError("Date must use YYYY-MM-DD")
date.fromisoformat(day)
~~~
This rejects strings such as **2026-02-30**, even though their shape is valid.
One intentional detail: conversions are **not** capped at clicks. Different conversion events and attribution methods can make that relationship more complicated. Imposing an unexamined conversion-to-click constraint could incorrectly reject legitimate-looking records.
These validations are part of the portfolio's fictional contract; actual platform-specific policies require separate research and source definitions.
## 4. Make file ingestion fail closed
Reading a CSV successfully does not mean the batch is safe to publish. The **normalize_files** function uses Python's csv.DictReader and checks for missing or duplicate headers, extra columns, empty inputs, malformed rows, and duplicate campaign-day keys.
The duplicate key is:
~~~python
key = (record["event_date"], source, record["campaign_id"])
if key in seen:
raise ValueError(
f"{source} line {line_number}: duplicate campaign-day {key}"
)
seen.add(key)
~~~
Including **source** in the key prevents campaign IDs from unrelated platforms from colliding. It also makes the expected grain explicit: one row per date, platform, and campaign in this demonstration.
Instead of silently picking one duplicate, the adapter fails the batch. That's a defensible starting point when no deterministic source precedence or revision timestamp exists.
After validating all input rows, it writes a temporary CSV **in the destination directory** and then calls **os.replace**:
~~~python
with tempfile.NamedTemporaryFile(
mode="w", newline="", encoding="utf-8",
dir=output.parent, prefix=".adapters-",
suffix=".tmp", delete=False,
) as handle:
writer = csv.DictWriter(handle, fieldnames=CANONICAL)
writer.writeheader()
writer.writerows(normalized)
os.replace(temp_name, output)
~~~
This is an abbreviated illustration of the actual code, which also sorts rows deterministically and cleans up the temporary file in a finally block.
Why write this way? A validation failure occurs **before** the existing output is replaced. A reader should not encounter a partially written final CSV during a successful same-filesystem replacement. This does not, by itself, provide database transactions, concurrent-writer coordination, or a crash-durability guarantee.
The [Python csv documentation](https://docs.python.org/3/library/csv.html) explains DictReader and DictWriter behavior.
## 5. Test failure paths, not just the happy path
The repository's [adapter test suite](https://github.com/rk268/retail-media-analytics-pipeline/blob/main/tests/test_source_adapters.py) includes cases for field mapping, micro-USD rounding, invalid dates, negative or non-integer metrics, missing columns, duplicate headers, and empty files.
My favorite regression case checks that a duplicate does **not overwrite the previous output**:
~~~python
output.write_text("previous output", encoding="utf-8")
with self.assertRaisesRegex(ValueError, "duplicate campaign-day"):
normalize_files({"google_ads": source}, output)
self.assertEqual(
output.read_text(encoding="utf-8"),
"previous output",
)
~~~
The test creates a source containing duplicate campaign-day rows before invoking the adapter. It validates a useful behavior, not just an exception message: a rejected batch leaves the existing artifact intact.
Another test loads one synthetic Google-like row and one Meta-like row, checks that both become the same canonical header, and verifies that their platform identities are preserved.
The relevant [October 24 commit](https://github.com/rk268/retail-media-analytics-pipeline/commit/d62a69e43981476f3d7f833a9bdaf0a59d05ce92) passed the repository's GitHub Actions CI workflow. A green CI result verifies the covered tests in that environment, not the correctness of real vendor integrations.
## 6. Connect adapters to the PySpark pipeline
The adapters use only Python's standard library. The output becomes input to the existing PySpark normalization stage:
~~~bash
python -m src.source_adapters \
--google data/raw/mock_google_ads.csv \
--meta data/raw/mock_meta_ads.csv \
--output data/raw/normalized_platform_exports.csv
spark-submit src/bronze_to_silver.py \
--input data/raw/normalized_platform_exports.csv \
--output data/silver/normalized_platform_exports
~~~
That separation is deliberate. Source-specific parsing, validation, and units belong at the ingestion boundary; reusable Spark transformations should operate on a stable contract.
For a small teaching dataset, collecting validated records in Python memory is reasonable. For millions of rows, I would move toward streaming or partitioned processing, bounded-memory validation, explicit batch identifiers, and durable staging.
## 7. What this still does not solve
**Currency and time zones:** The mock exports assume USD and aligned reporting dates. Production adapters need explicit currency codes, exchange-rate policies, account time zones, and late-arriving data rules.
**Measurement semantics:** A link click is not automatically equivalent to every platform's click metric; purchases and conversions may use different attribution windows. Matching column names does not establish comparable KPIs.
**Operational ingestion:** This module does not call vendor APIs, paginate results, authenticate, maintain incremental checkpoints, or quarantine invalid rows. Those are future engineering tasks.
**Data revisions:** If a platform restates yesterday's campaign metrics, a production pipeline needs idempotent upserts and revision-aware reconciliation rather than treating every repeated campaign-day as an unconditional error.
## Takeaway
A robust multi-source pipeline starts with a **documented contract**, explicit unit conversions, failure-aware file handling, and tests that verify what happens when inputs are wrong. The Spark transformations are only as trustworthy as the boundary feeding them.
You can inspect the [source code](https://github.com/rk268/retail-media-analytics-pipeline/blob/main/src/source_adapters.py), [tests](https://github.com/rk268/retail-media-analytics-pipeline/blob/main/tests/test_source_adapters.py), and [adapter design notes](https://github.com/rk268/retail-media-analytics-pipeline/blob/main/docs/source-adapters.md).
**Question:** When two platforms disagree on a metric's definition, do you normalize to a common semantic layer immediately, or retain both native and standardized measures for auditability?
Top comments (0)