DEV Community: Rupak Kulkarni

A Green Spark Job Isn't Enough: Building a Fail-Closed Pipeline Health Monitor in Python

Rupak Kulkarni — Mon, 02 Nov 2026 15:03:33 +0000

A Spark pipeline can produce the right files and still be operationally unhealthy. Maybe the Silver stage failed once and recovered. Maybe a process started but never wrote its completion event. Or perhaps a dashboard reads a half-written health report and displays a misleading green status.

For a retail-media analytics pipeline, those distinctions matter: delayed campaign metrics can be mistaken for a drop in performance, while an incomplete Gold refresh can look like a successful reporting cycle.

I recently added a Python-only operational monitor to my public, synthetic-data retail-media analytics pipeline. This article walks through the actual design, its event contract, failure semantics, tests, and limitations. The monitoring implementation, orchestrator, and tests are public. All examples here use synthetic events, not employer or production advertising data.

1. Separate pipeline execution from health interpretation

The demonstration pipeline has two ordered PySpark stages:

Synthetic campaign CSV
       |
       v
Stage 1: Bronze -> Silver (PySpark)
       |
       v
Stage 2: Silver -> Gold (PySpark)
       |
       v
Reporting-ready metrics

The local orchestrator invokes spark-submit for each stage, stops if a stage exhausts its retries, and appends structured events. The monitor is a separate reader: it does not launch Spark or decide which command to retry. It reconstructs the current health of observed runs from the events already written.

This separation has a practical benefit. A lightweight monitoring job can run without starting a Spark session, and a dashboard can consume its output without parsing free-form executor logs.

2. Use an event contract, not keyword searches

The orchestrator writes one JSON object per line to data/outputs/pipeline_events.jsonl. Every event includes a run identifier, stage number, attempt number, status, and timezone-aware UTC timestamp. Terminal events also carry a return code; failures may include bounded error text.

For example, these are illustrative synthetic events from a stage that fails once and then succeeds:

{"run_id":"demo-001","stage":1,"attempt":1,"status":"started","timestamp_utc":"2026-10-31T12:00:00+00:00"}
{"run_id":"demo-001","stage":1,"attempt":1,"status":"failed","returncode":1,"error":"temporary failure","timestamp_utc":"2026-10-31T12:00:03+00:00"}
{"run_id":"demo-001","stage":1,"attempt":2,"status":"started","timestamp_utc":"2026-10-31T12:00:05+00:00"}
{"run_id":"demo-001","stage":1,"attempt":2,"status":"succeeded","returncode":0,"timestamp_utc":"2026-10-31T12:00:08+00:00"}
{"run_id":"demo-001","stage":2,"attempt":1,"status":"started","timestamp_utc":"2026-10-31T12:00:09+00:00"}
{"run_id":"demo-001","stage":2,"attempt":1,"status":"succeeded","returncode":0,"timestamp_utc":"2026-10-31T12:00:14+00:00"}

JSON Lines works well here because events can be appended incrementally and processed line by line. See the JSON Lines format conventions.

Notice the distinction between an attempt and a run. A failed attempt does not necessarily mean a failed run: the next attempt may recover. Conversely, one successful stage does not make a two-stage pipeline successful.

3. Validate before declaring anything healthy

The monitor's summarize_events() function checks the event stream instead of assuming it is trustworthy. Among other rules, it rejects:

Missing or invalid run IDs, stage numbers, attempts, or statuses.
Naive timestamps without timezone information and per-run events arriving out of chronological order.
Duplicate attempt starts, skipped attempt numbers, and retries without a preceding failed attempt.
Terminal events without an active start, or success events with a nonzero return code.
Starting stage 2 before stage 1 has succeeded.

A malformed log raises an error rather than being quietly treated as an empty or successful run. That is a deliberate fail-closed policy: a green dashboard should require positive evidence.

The run-level state is calculated from the latest state of each expected stage. A run is succeeded only when every expected stage succeeded. Otherwise, a failed stage takes precedence, followed by a stalled stage, a currently running stage, or an incomplete run. The output also retains attempt counts, retry counts, failed-attempt counts, and the last bounded error for each stage.

4. Detect stalls without confusing them with failures

A process can emit started and then stop producing events. The monitor classifies an unfinished attempt as stalled only when its age exceeds a configurable threshold:

# Excerpted logic from src/pipeline_monitor.py
if stage_status == "running" and (
    current - last["started"]
).total_seconds() > stale_after_seconds:
    stage_status = "stalled"

The default threshold is 600 seconds. A recently started attempt remains running; an old completed attempt remains succeeded, not stalled. The code uses timezone-aware timestamps and permits a fixed now argument, which makes boundary cases deterministic in tests.

There is an important caveat: a stall is inferred from the absence of a completion event, not proven process death. Long legitimate jobs need an appropriate threshold or a heartbeat protocol. A monitoring snapshot can also change if later retry events arrive.

5. Export dashboard JSON atomically

The monitor produces a snapshot with schema_version, generated_at_utc, latest_run_id, run_count, per-status health_counts, and detailed run/stage records. A consumer could load that JSON into a small operational dashboard or Power BI dataflow; the repository does not claim a deployed production dashboard.

Rather than writing directly over the existing report, the implementation writes a temporary file in the destination directory, flushes and syncs it, then replaces the destination with os.replace():

# Simplified illustration of the repository's write pattern
with tempfile.NamedTemporaryFile(
    mode="w", encoding="utf-8", dir=target.parent,
    delete=False,
) as stream:
    json.dump(summary, stream, allow_nan=False)
    stream.flush()
    os.fsync(stream.fileno())
    temp_path = Path(stream.name)

os.replace(temp_path, target)

The real implementation also creates the destination directory and cleans up temporary files in a finally block. Keeping the temporary file in the same directory supports atomic replacement on the same filesystem; see Python's os.replace documentation. If event validation fails, the previous dashboard file is preserved rather than replaced with a misleading partial report.

6. Make the monitor useful in automation

After the sample pipeline has generated events, run:

python -m src.pipeline_monitor \
  --events data/outputs/pipeline_events.jsonl \
  --output data/outputs/pipeline_health.json \
  --expected-stages 2 \
  --stale-after-seconds 600 \
  --fail-on-unhealthy

With --fail-on-unhealthy, the exit codes are intentionally meaningful:

Exit code	Meaning
`0`	Every observed run succeeded.
`1`	Empty log or at least one failed, stalled, running, or incomplete run.
`2`	Invalid events or an input/output error.

Without --fail-on-unhealthy, an otherwise valid snapshot can be written with exit code 0 even if it describes unhealthy runs. That distinction matters when integrating the command into a scheduled check or a CI step. GitHub Actions uses process exit codes to determine success or failure; wiring this monitor into a deployment gate would be a separate integration decision, not something this example claims is already configured.

7. Test state transitions, not only happy paths

The monitor tests use synthetic JSONL files, temporary directories, and a fixed UTC clock. They cover successful two-stage runs, retry-then-success, permanently failed stages, running versus stalled attempts, invalid event ordering, orphan completions, invalid return codes, empty logs, atomic output, and unhealthy CLI exits.

Run the dedicated tests with:

python -m unittest discover -s tests -p 'test_pipeline_monitor.py' -v

The October 31 implementation commit added the monitor, documentation, and its tests. The corresponding GitHub Actions run passed. That validates this repository's tested behavior; it is not a claim about production reliability or benchmark performance.

8. What this design does not solve

Operational health is not data correctness. A pipeline can finish every stage while still producing bad spend, impressions, or conversion metrics; separate data-quality and KPI checks remain necessary. The monitor also does not provide distributed locks, a durable event broker, multi-worker coordination, heartbeat-based liveness, alert routing, or log retention policies. The JSONL stream can contain command arguments and error text, so production logs would need secret redaction and access controls.

For a production system, I would add run ownership, heartbeats, retention, idempotent event ingestion, structured alerts, and a clear policy for late or out-of-order events before using these snapshots as an operational source of truth.

Takeaway: A trustworthy green status is something you prove from ordered, validated evidence—not something you infer because the last command printed “done.” A small, tested Python monitor can make retries, partial runs, and stalled stages visible without turning every health check into another Spark job.

Code and further reading: Repository · Monitoring guide · JSON Lines · Python atomic replacement

From Google- and Meta-Like CSVs to a Trusted Schema: Building Fail-Closed Python Adapters

Rupak Kulkarni — Mon, 26 Oct 2026 14:34:37 +0000

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?

Testing PySpark Data Quality in CI: A Practical Retail Media Example

Rupak Kulkarni — Wed, 21 Oct 2026 19:30:59 +0000

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?**

Retail Media Incrementality in Python: What a Synthetic Holdout Can (and Cannot) Tell Us

Rupak Kulkarni — Sun, 18 Oct 2026 13:50:51 +0000

I recently added a synthetic incrementality example to my public Retail Media Analytics Pipeline. It gave me an opportunity to explore a distinction that is easy to lose in advertising dashboards:

Attributed conversions are not the same as incremental conversions.

A report can tell us how many conversions were associated with an ad campaign. The harder question is: how many of those outcomes happened because of the campaign, rather than happening anyway?

This tutorial walks through a fabricated treatment/control example, shows how to calculate basic lift in Python, and explains the assumptions that would be required for a real causal experiment.

Transparency: All values below are synthetic. The code assumes a randomized experiment; it does not prove random assignment happened, and it does not establish any real-world advertising impact.

Why ROAS and incrementality answer different questions

In a conventional campaign report, return on ad spend (ROAS) is often calculated as:

ROAS = attributed revenue / advertising spend

That is a useful descriptive metric, but the numerator depends on an attribution methodology. Attribution isn't automatically causal identification.

An incrementality study instead compares outcomes for people eligible to receive an intervention with outcomes for a comparable group that did not receive it.

The simplest conceptual design has two arms:

Treatment: eligible users assigned to the advertising intervention.
Control: comparable users assigned to a holdout.

For this comparison to support a causal interpretation, assignment and exposure must be handled correctly. Aggregated campaign reports alone cannot establish that.

A small, intentionally synthetic dataset

The example added to the repository stores aggregate cohort counts in data/raw/simulated_holdout.csv:

group,users,conversions,revenue,spend
treatment,50000,3000,240000.00,45000.00
control,50000,2600,208000.00,0.00

Both groups have 50,000 users. Treatment has 3,000 conversions; control has 2,600.

These numbers are deliberately simple, so the calculations are easy to inspect. They are not from an employer, an ad platform, or an actual experiment.

Step 1: Compare conversion rates

Conversion rate is the number of conversions divided by users in that arm.

treatment_rate = 3000 / 50000
control_rate = 2600 / 50000

absolute_lift = treatment_rate - control_rate
relative_lift = absolute_lift / control_rate

print(treatment_rate)  # 0.060 = 6.0%
print(control_rate)    # 0.052 = 5.2%
print(absolute_lift)   # 0.008 = 0.8 percentage points
print(relative_lift)   # approx. 15.38%

The distinction between percentage points and percent matters:

Treatment conversion rate: 6.0%
Control conversion rate: 5.2%
Absolute difference: 0.8 percentage points
Relative lift: approximately 15.4%

A chart titled “15.4% incremental lift” without specifying the denominator can be misleading.

Step 2: Estimate incremental conversions at treatment scale

In this demonstration, incremental conversions are the conversion-rate difference multiplied by the number of treated users:

incremental_conversions = absolute_lift * 50000
print(incremental_conversions)  # 400.0

The estimate is 400 incremental conversions, conditional on the causal assumptions of the hypothetical experiment.

Scaling to treatment size is important when treatment and control groups are unequal. Simply subtracting raw conversion counts across differently sized cohorts would conflate group size with performance.

Step 3: Estimate incremental revenue and iROAS

The code works with revenue per user rather than just subtracting totals. For equally sized groups, the result is numerically the same here:

treatment_rpu = 240000 / 50000
control_rpu = 208000 / 50000

incremental_revenue = (treatment_rpu - control_rpu) * 50000
incremental_spend = 45000 - 0

iroas = incremental_revenue / incremental_spend

print(incremental_revenue)  # 32000.0
print(iroas)                # approx. 0.71

That produces a $32,000 estimated incremental revenue difference and iROAS of approximately 0.71 in the toy dataset.

Note that attributed ROAS and iROAS can differ substantially. They measure different things.

Also, this example does not include uncertainty estimates for revenue or iROAS. A point estimate should never be presented as definitive proof of impact.

Step 4: Add an approximate uncertainty interval

The repository calculates a large-sample, unpooled Wald approximation for the difference in proportions:

import math

pt, pc = 0.06, 0.052
nt, nc = 50000, 50000

difference = pt - pc
se = math.sqrt(
    pt * (1 - pt) / nt
    + pc * (1 - pc) / nc
)
ci = (difference - 1.96 * se, difference + 1.96 * se)

For these hypothetical counts, the approximate 95% interval for the conversion-rate difference is about 0.52 to 1.08 percentage points.

That interval covers the conversion-rate difference, not incremental revenue or iROAS. The Wald method can perform poorly with small samples or rare events; production analysis may require a better interval and a design-aware variance estimator.

A numerically narrow confidence interval does not resolve poor randomization, interference, invalid conversion definitions, or mismatched observation windows.

Step 5: Make the analysis reproducible

The repository includes a standalone module, src/incrementality.py, which:

Validates that exactly one treatment and one control row are provided.
Rejects negative or nonfinite monetary amounts.
Rejects conversion counts that exceed user counts.
Calculates conversion-rate lift, treatment-scaled incremental conversions and revenue, and incremental ROAS.
Writes results to JSON for inspection.

Run it from the repository root:

python -m src.incrementality \
  --input data/raw/simulated_holdout.csv \
  --output data/outputs/incrementality_summary.json

It also includes seven unit tests, covering calculations and invalid inputs:

python -m unittest discover -s tests -p 'test_incrementality.py' -v

The GitHub Actions CI runs for this addition passed on October 17, 2026. This is a small educational module, not a production experiment framework.

What I'd add before calling this production-ready

A real incrementality platform needs far more than a CSV and a calculation function.

Randomization evidence. Record assignment at the relevant unit (user, household, store, region, or another cluster), and detect sample-ratio mismatches.

Measurement consistency. Align conversion definitions, time windows, exclusions, identity resolution, and delayed outcomes.

Interference and spillover. Exposure in the treatment group must not silently alter outcomes in the control group, or vice versa.

Uncertainty for money metrics. Revenue often has a skewed distribution. A credible design should estimate uncertainty for incremental revenue and iROAS rather than reporting only point estimates.

Operational checks. Pre-register primary outcomes, inspect data completeness, document assignment and analysis versions, and enforce appropriate privacy controls.

An especially important caveat: this project's randomized treatment/control design is an assumption of the synthetic exercise. The CSV has no assignment logs, so the code cannot independently validate that assumption.

What I learned from building the example

Three principles stand out:

Separate descriptive and causal metrics. Cross-channel campaign dashboards tell us useful things, but they do not automatically measure causal lift.
Write assumptions next to the numbers. “400 incremental conversions” is incomplete without treatment size, experiment validity, and uncertainty.
Make toy models auditable. Simple synthetic examples, unit tests, and readable JSON outputs can clarify the mathematics before a more sophisticated system is attempted.

You can explore the complete incrementality walkthrough, Python implementation, and tests.

Question for the community: What tends to cause more trouble in real incrementality programs—assignment integrity, conversion measurement, or communicating uncertainty to stakeholders?

— Rupak Kulkarni | Data Engineering, PySpark, Retail Media Analytics

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

Rupak Kulkarni — Thu, 15 Oct 2026 15:41:42 +0000

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:

Missing required columns
Missing or blank business keys
Null or negative metric values
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?