Building a Retail Media Analytics Pipeline with PySpark: Bronze, Silver, and Gold

A retail-media dashboard can look straightforward: impressions, clicks, spend, conversions, and revenue by campaign. But the engineering underneath becomes more interesting when those metrics originate from several advertising platforms.

To explore that problem, I put together a public portfolio project using PySpark, SQL, and the medallion architecture. It works with simulated campaign data representing Google Ads, Meta Ads, Pinterest, and Snapchat—no employer data or confidential datasets.

Repository: Retail Media Analytics Pipeline on GitHub

In this walkthrough, I'll show how the pipeline is structured, how the KPI layer works, and why data-quality checks matter before aggregating campaigns.

The problem: one dashboard, multiple definitions

Suppose an analyst wants to compare campaign performance across platforms. At a minimum, they need a common way to describe:

• When the activity happened
• Which platform and campaign generated it
• Impressions and clicks
• Spend, conversions, and attributed revenue

A shared schema makes the engineering consistent, but doesn't magically make measurement definitions identical. Attribution windows, conversion events, currency, and reporting time zones can differ by platform. That's why a unified data model is a starting point—not the finish line.

For this demo, the input is intentionally standardized:

event_date, platform, campaign_id, campaign_name,
impressions, clicks, spend, conversions, revenue

In a production system, I'd first implement adapters for each native platform export or API. Those adapters are a planned extension of this repository, not something I'm claiming is complete today.

Architecture: Bronze → Silver → Gold

The repo follows a familiar medallion pattern:

Simulated platform extracts (CSV)
              |
              v
         BRONZE / raw
              |
              v
        SILVER / clean
     schema + normalization
              |
              v
        GOLD / analytics
   campaign aggregates + KPIs
              |
          SQL / BI

Bronze: preserve what arrived

The sample data lives under data/raw/. In a production design, Bronze should retain the original source records with ingestion timestamps, batch identifiers, and source metadata. That gives you a reproducible audit trail if a platform changes its schema.

Silver: standardize data types and identifiers

The current src/bronze_to_silver.py implementation selects expected columns and normalizes the records with PySpark. Among other things, it:

• Parses event_date into a date
• Lowercases and trims the platform identifier
• Trims campaign identifiers and names
• Casts count fields to long and monetary measures to double
• Removes duplicates by date, platform, and campaign ID
• Filters records that fail selected business rules

For example, the relevant Spark expressions look like this:

from pyspark.sql import functions as F, types as T

clean = (
    df.withColumn("event_date", F.to_date("event_date"))
      .withColumn("platform", F.lower(F.trim("platform")))
      .withColumn("campaign_id", F.trim("campaign_id"))
      .withColumn("impressions", F.col("impressions").cast(T.LongType()))
      .withColumn("spend", F.col("spend").cast(T.DoubleType()))
)

The sample pipeline writes partitioned Parquet output. The important next step is to quarantine rejected rows instead of only filtering them, so errors remain inspectable. That's on the project's roadmap.

Gold: turn observations into usable KPIs

The implementation in src/silver_to_gold.py groups records by platform, campaign_id, and campaign_name, then sums impressions, clicks, spend, conversions, and revenue.

Ratios should be calculated after aggregation, rather than averaging each input row's rate.

The code uses a safe-divide expression:

from pyspark.sql import functions as F

def safe_divide(numerator, denominator):
    return F.when(
        denominator > 0,
        numerator / denominator
    ).otherwise(F.lit(0.0))

Then it derives metrics such as:

KPI	Calculation
CTR	Clicks / impressions
CPC	Spend / clicks
CPM	Spend × 1,000 / impressions
Conversion rate	Conversions / clicks
CPA	Spend / conversions
ROAS	Revenue / spend

For a hypothetical campaign with 10,000 impressions, 500 clicks, $250 of spend, 25 conversions, and $1,000 in attributed revenue:

• CTR = 5%
• CPC = $0.50
• CPM = $25
• Conversion rate = 5%
• CPA = $10
• ROAS = 4.0

This example is illustrative—not a measured marketing result. Also, representing an undefined ratio as 0.0 is a conscious implementation choice; for some analytics products, NULL plus a missing-denominator flag would communicate that distinction more accurately.

Add data quality before trusting a dashboard

A pipeline can finish successfully and still produce misleading numbers. Imagine a row with negative spend, missing conversions, or more clicks than impressions. If it reaches a dashboard without a warning, someone may interpret those values as business performance.

The repository includes a reusable PySpark quality-report module. It checks for:

1. Missing required columns
2. Missing or blank business keys
3. Null or negative metric values
4. Clicks greater than impressions

Rather than silently removing problematic rows, the quality report aggregates the counts of rule violations. A companion function fails the check when violations are present:

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

The project includes tests with synthetic good and bad records. One important distinction: this reusable gate exists and has tests, but it is not yet wired into every step of the command-line processing flow. Integrating it with an explicit reject/quarantine path is further work.

CI is part of the pipeline, too

Tests are useful only if the build environment can run them. The repository's GitHub Actions workflow installs Java 17 and the Python dependencies declared in requirements.txt before invoking python -m pytest -q.

That's a practical lesson for any Spark portfolio or production project: runtime dependencies and reproducible tests belong in version control, alongside transformation code.

You can review the CI workflow and test cases directly.

What this example does not solve yet

The repository is a compact, public learning project, not a production-grade retail-media measurement platform. Still on the roadmap:

• Native schema adapters for individual advertising platforms
• Reject/quarantine datasets and richer quality expectations
• Incremental ingestion, idempotent processing, and backfills
• Orchestration and operational monitoring
• Attribution-window harmonization and incrementality examples
• A dashboard-ready semantic layer

Those details matter. Two platforms can report the same numerical ROAS while using different definitions of a conversion or revenue. Data engineering should make such differences explicit instead of hiding them behind a neat chart.

Takeaway

I think of the medallion approach here as a sequence of questions:

• Bronze: What exactly arrived?
• Silver: Can the data be trusted and compared technically?
• Gold: Which business metrics are ready to consume?

A useful first version doesn't need to solve every measurement challenge. It needs a clear data model, understandable transformations, honest limitations, and tests that protect against obvious errors.

Explore the code: github.com/rk268/retail-media-analytics-pipeline.

If you've built cross-channel reporting pipelines, what has been more challenging in practice: schema normalization, metric definitions, or attribution differences?