A pipeline can complete successfully and still deliver a misleading answer. In retail-media analytics, negative spend, missing conversions, or more clicks than impressions can distort campaign KPIs before the first dashboard refresh.
I explored this problem in a public PySpark retail-media analytics portfolio using synthetic campaign records only. This walkthrough covers the actual data-quality implementation, pytest checks, GitHub Actions configuration, and a recent improvement that enforces validation before producing Gold-layer metrics.
The repository is a demonstration, not a claim about live advertiser data or a production deployment.
1. Define a data contract
The quality module starts by declaring expected columns and business keys:
REQUIRED = (
"event_date", "platform", "campaign_id",
"impressions", "clicks", "spend",
"conversions", "revenue",
)
METRICS = ("impressions", "clicks", "spend", "conversions", "revenue")
KEYS = ("event_date", "platform", "campaign_id")
~~~{% endraw %}
The checks look for missing required columns, blank or missing keys, null or negative metrics, and click counts that exceed impressions. These rules do not prove causal accuracy or reconcile attribution windows; they protect a narrower, essential contract.
## 2. Build reusable Spark quality checks
Column expressions work across DataFrames without looping over records in Python. The implementation combines metric-level checks such as:{% raw %}
~~~python
from pyspark.sql import functions as F
negative = F.lit(False)
for metric in METRICS:
negative = negative | (F.col(metric) < 0)
~~~{% endraw %}
The function quality_report(df) returns a single-row report containing the total number of observations and violation counts by rule.
**A useful distinction:** rule-violation counts are not counts of unique invalid records. One row can have negative spend *and* more clicks than impressions. Adding those violation counts together would double-count the affected row.
The companion function assert_quality(df) converts that report into an enforcement decision:{% raw %}
~~~python
def assert_quality(df):
counts = quality_report(df).first().asDict()
failures = {
key: value
for key, value in counts.items()
if key != "row_count" and value
}
if failures:
raise ValueError(
f"Retail media data-quality gate failed: {failures}"
)
return counts
~~~{% endraw %}
This separates diagnosis from policy: a report tells you what went wrong; a gate decides whether downstream processing is allowed to proceed.
## 3. Test the rules with synthetic counterexamples
Small, intentionally invalid DataFrames are excellent tests for a quality gate. The [pytest cases](https://github.com/rk268/retail-media-analytics-pipeline/blob/main/tests/test_data_quality.py) include a record with negative spend and excessive clicks, as well as one with a blank platform and missing conversion count.{% raw %}
~~~python
df = rows(spark, [
("2026-10-01", "google", "c1", 10, 12, -1.0, 2, 4.0),
("2026-10-02", " ", "c2", 20, 2, 1.0, None, 5.0),
])
report = quality_report(df).first().asDict()
assert report["row_count"] == 2
assert report["negative_metrics"] == 1
assert report["clicks_exceed_impressions"] == 1
assert report["invalid_keys"] == 1
assert report["null_metrics"] == 1
~~~{% endraw %}
The project also tests valid data, missing schema fields, and expected exceptions. Verifying that good rows pass matters just as much as showing that bad rows fail.
Tests use a module-scoped fixture to share one local Spark session:{% raw %}
~~~python
import pytest
from pyspark.sql import SparkSession
@pytest.fixture(scope="module")
def spark():
session = SparkSession.builder.master("local[1]").appName(
"quality-tests"
).getOrCreate()
yield session
session.stop()
~~~{% endraw %}
A local session avoids requiring a separate running cluster for these small examples. Explicit schemas make the tests more predictable when values are missing or nullable.
## 4. Connect the gate to the Gold build
A reusable quality function is not enough unless the transformation path actually invokes it.
The [Silver-to-Gold builder](https://github.com/rk268/retail-media-analytics-pipeline/blob/main/src/silver_to_gold.py) now calls the gate before aggregating campaigns:{% raw %}
~~~python
from src.data_quality import assert_quality
def build_gold(df):
"""Reject invalid Silver records before publishing Gold aggregates."""
assert_quality(df)
aggregated = (
df.groupBy("platform", "campaign_id", "campaign_name")
.agg(
F.sum("impressions").alias("impressions"),
F.sum("clicks").alias("clicks"),
F.sum("spend").alias("spend"),
F.sum("conversions").alias("conversions"),
F.sum("revenue").alias("revenue"),
F.min("event_date").alias("start_date"),
F.max("event_date").alias("end_date"),
)
)
# Additional KPI columns are derived from aggregated totals.
~~~{% endraw %}
The trade-off is explicit: the current Gold build **fails closed** when input data violates the contract. It does not yet write a quarantine dataset containing rejected rows. That remains an improvement to implement.
## 5. Test the actual KPI calculations
Data quality is only one part of correctness. The new [Gold-layer tests](https://github.com/rk268/retail-media-analytics-pipeline/blob/main/tests/test_silver_to_gold_quality.py) also confirm that a campaign with 100 impressions and 10 clicks, followed by 300 impressions and 30 clicks, aggregates to 400 impressions and 40 clicks.
That implies CTR = 40 / 400 = **10%**. Ratios should be computed after summing numerator and denominator, rather than by naively averaging row-level rates.
The regression tests also check CPC, CPM, CPA, ROAS, negative spend, blank business keys, and zero-denominator behavior.
For undefined denominators, the project's safe-divide expression returns 0.0. That is an implementation decision; other BI products may prefer a null result and a flag explaining why the ratio could not be calculated.
## 6. Make quality checks part of CI
The [GitHub Actions CI workflow](https://github.com/rk268/retail-media-analytics-pipeline/blob/main/.github/workflows/ci.yml) configures Java 17, Python 3.11, project dependencies, and this command:{% raw %}
~~~bash
python -m pytest -q
~~~
The project's latest quality-gate update has a [successful GitHub Actions run](https://github.com/rk268/retail-media-analytics-pipeline/actions/runs/39661716424). This is evidence that the covered rules and transformation tests passed in that configured build environment.
It does **not** prove that every reporting assumption is correct. Data from different advertising platforms can still reflect incompatible currencies, conversion definitions, time zones, or attribution windows.
## 7. The next engineering improvements
**Empty data feeds:** Is a zero-row input an acceptable quiet day or evidence of an upstream failure? Add an explicit policy and a test.
**Quarantined records:** Bronze-to-Silver currently filters invalid records. Preserve rejects with failure reasons, batch identifiers, and source metadata instead of allowing silent loss.
**Deterministic deduplication:** When the same date, platform, and campaign appear more than once with conflicting metrics, decide which record wins or raise an error.
**Measurement semantics:** Harmonize the definitions of spend, conversion, and attribution before treating cross-channel numbers as equivalent.
These are future improvements, not features I am claiming already exist.
## What I learned
A practical validation strategy has several layers: declare the schema, test bad records deliberately, report violations, enforce a clear gate, and run the same checks automatically on every change.
The most valuable outcome is not a green badge by itself. It is a reliable, inspectable link between the business assumptions in your data contract and the tests that guard those assumptions.
**Code and tests:** [Retail Media Analytics Pipeline](https://github.com/rk268/retail-media-analytics-pipeline)
### Further reading
- [Apache Spark: Testing PySpark applications](https://spark.apache.org/docs/3.5.9/api/python/getting_started/testing_pyspark.html)
- [pytest: How fixtures work](https://docs.pytest.org/en/stable/how-to/fixtures.html)
- [GitHub Docs: Building and testing Python](https://docs.github.com/en/actions/tutorials/build-and-test-code/python)
**How do you handle invalid records in your data pipelines: fail the whole batch, quarantine them and continue, or use a threshold-based policy?**
Top comments (0)