Clean a Marketing Dataset With Python and Pandas

Know what one row means.

campaigns

Key · campaign_id

One row per campaign.

daily_spend

Key · spend_id

One row per campaign and active date.

leads

Key · lead_id

One row per attributed lead.

conversions

Key · conversion_id

One row per converted lead.

channel_performance

Key · channel

One row per campaign channel after spend and conversions are aggregated separately.

Fix the meaning before the code.

Valid source record

A row with its documented required values, a parseable date where applicable, and numeric measures that can be converted without guessing.

Duplicate source record

A repeated primary identifier in one source table. The challenge intentionally duplicates at least one complete record in every table.

Attributed conversion count

The number of cleaned conversion records connected to leads and their attributed campaigns. It is not an incrementality estimate.

Work from source grain to tested output.

1. 01

   Load every released challenge table

   Extract the published challenge ZIP, preserve identifiers as text, and profile each table before changing it.

   from pathlib import Path
   
   import pandas as pd
   
   ROOT = Path("marketing-campaigns-v1.0.0-cleaning-challenge")
   RAW = ROOT / "raw"
   
   table_keys = {
       "campaigns": "campaign_id",
       "daily_spend": "spend_id",
       "leads": "lead_id",
       "conversions": "conversion_id",
   }
   
   tables = {
       name: pd.read_csv(
           RAW / f"{name}.csv",
           dtype="string",
           keep_default_na=False,
       )
       for name in table_keys
   }
   
   for name, frame in tables.items():
       print(name, frame.shape)
       print(frame.head(3))

   Verification

   • The extracted package contains raw/campaigns.csv, raw/daily_spend.csv, raw/leads.csv, and raw/conversions.csv.
   • Identifiers retain their prefixes and leading zeros.
   • The original extracted files remain unchanged.
2. 02

   Normalize documented issues and preserve rejected rows

   Apply explicit rules for whitespace, labels, dates, and numeric types; quarantine missing required values; then remove duplicate primary keys.

   required = {
       "campaigns": [
           "campaign_id", "campaign_name", "channel", "objective",
           "start_date", "end_date", "budget",
       ],
       "daily_spend": [
           "spend_id", "campaign_id", "spend_date",
           "impressions", "clicks", "spend",
       ],
       "leads": [
           "lead_id", "campaign_id", "created_at",
           "lead_source", "lead_score",
       ],
       "conversions": [
           "conversion_id", "lead_id", "converted_at",
           "conversion_type", "revenue",
       ],
   }
   date_columns = {
       "campaigns": ["start_date", "end_date"],
       "daily_spend": ["spend_date"],
       "leads": ["created_at"],
       "conversions": ["converted_at"],
   }
   numeric_columns = {
       "campaigns": ["budget"],
       "daily_spend": ["impressions", "clicks", "spend"],
       "leads": ["lead_score"],
       "conversions": ["revenue"],
   }
   category_columns = {
       "campaigns": ["channel", "objective"],
       "daily_spend": [],
       "leads": ["lead_source"],
       "conversions": ["conversion_type"],
   }
   
   clean_tables = {}
   quarantine_parts = []
   
   for name, frame in tables.items():
       work = frame.apply(lambda column: column.str.strip()).replace("", pd.NA)
       work.insert(0, "source_row", range(2, len(work) + 2))
   
       if name == "campaigns":
           work["campaign_name"] = work["campaign_name"].str.title()
       for column in category_columns[name]:
           work[column] = work[column].str.lower()
       for column in date_columns[name]:
           normalized = work[column].str.replace("/", "-", regex=False)
           work[column] = pd.to_datetime(
               normalized,
               format="%Y-%m-%d",
               errors="coerce",
           )
       for column in numeric_columns[name]:
           work[column] = pd.to_numeric(work[column], errors="coerce")
   
       missing_required = work[required[name]].isna().any(axis=1)
       duplicate_key = work.duplicated(table_keys[name], keep="first")
   
       issue = pd.Series("missing required value", index=work.index)
       issue.loc[duplicate_key] = "duplicate primary key"
       rejected = work.loc[missing_required | duplicate_key].copy()
       rejected["source_table"] = name
       rejected["issue"] = issue.loc[rejected.index]
       quarantine_parts.append(rejected)
   
       clean_tables[name] = (
           work.loc[~missing_required & ~duplicate_key]
           .drop(columns="source_row")
           .reset_index(drop=True)
       )
   
   quarantine = pd.concat(quarantine_parts, ignore_index=True)
   quarantine.to_csv("marketing-cleaning-quarantine-v1.0.0.csv", index=False)
   
   Path("clean").mkdir(exist_ok=True)
   for name, frame in clean_tables.items():
       frame.to_csv(f"clean/{name}.csv", index=False)

   Verification

   • Every rejected row retains its source table, source row, and issue.
   • Slash-delimited challenge dates are normalized before strict parsing.
   • Every cleaned table is unique on its published primary identifier.
3. 03

   Aggregate compatible grains and test the result

   Aggregate campaign spend and attributed conversions separately before combining them at campaign and channel grain.

   campaigns = clean_tables["campaigns"]
   daily_spend = clean_tables["daily_spend"]
   leads = clean_tables["leads"]
   conversions = clean_tables["conversions"]
   
   valid_campaigns = set(campaigns["campaign_id"])
   valid_leads = set(leads["lead_id"])
   
   orphan_spend = daily_spend.loc[
       ~daily_spend["campaign_id"].isin(valid_campaigns)
   ]
   orphan_leads = leads.loc[
       ~leads["campaign_id"].isin(valid_campaigns)
   ]
   orphan_conversions = conversions.loc[
       ~conversions["lead_id"].isin(valid_leads)
   ]
   
   campaign_spend = (
       daily_spend.loc[daily_spend["campaign_id"].isin(valid_campaigns)]
       .groupby("campaign_id", as_index=False)
       .agg(spend=("spend", "sum"))
   )
   
   campaign_conversions = (
       conversions.loc[conversions["lead_id"].isin(valid_leads)]
       .merge(
           leads[["lead_id", "campaign_id"]],
           on="lead_id",
           how="inner",
           validate="one_to_one",
       )
       .loc[lambda frame: frame["campaign_id"].isin(valid_campaigns)]
       .groupby("campaign_id", as_index=False)
       .agg(conversions=("conversion_id", "count"))
   )
   
   campaign_performance = (
       campaigns[["campaign_id", "channel"]]
       .merge(campaign_spend, on="campaign_id", how="left", validate="one_to_one")
       .merge(
           campaign_conversions,
           on="campaign_id",
           how="left",
           validate="one_to_one",
       )
       .fillna({"spend": 0, "conversions": 0})
   )
   
   channel_performance = (
       campaign_performance.groupby("channel", as_index=False)
       .agg(spend=("spend", "sum"), conversions=("conversions", "sum"))
       .sort_values("channel")
   )
   channel_performance["cost_per_conversion"] = channel_performance[
       "spend"
   ].div(channel_performance["conversions"].replace(0, pd.NA))
   
   channel_performance.to_csv(
       "marketing_channel_performance_v1.0.0.csv",
       index=False,
   )

   for name, frame in clean_tables.items():
       assert not frame[table_keys[name]].duplicated().any()
       assert frame[required[name]].notna().all().all()
   
   assert daily_spend["impressions"].ge(0).all()
   assert daily_spend["clicks"].ge(0).all()
   assert daily_spend["spend"].ge(0).all()
   assert leads["lead_score"].between(1, 100).all()
   assert conversions["revenue"].ge(0).all()
   assert not channel_performance["channel"].duplicated().any()
   assert abs(
       campaign_spend["spend"].sum()
       - daily_spend.loc[
           daily_spend["campaign_id"].isin(valid_campaigns),
           "spend",
       ].sum()
   ) < 0.01
   
   print(
       {
           "quarantined_rows": len(quarantine),
           "orphan_spend_rows": len(orphan_spend),
           "orphan_lead_rows": len(orphan_leads),
           "orphan_conversion_rows": len(orphan_conversions),
       }
   )

   Verification

   • All assertions pass for the v1.0.0 cleaning-challenge variant.
   • Rows whose parent was quarantined are reported instead of silently joined.
   • A channel with zero conversions retains a missing cost per conversion instead of an infinite value.

What this result does not claim.

• —The walkthrough uses the first 2,000 rows supplied for each challenge table, not the complete clean release.
• —The challenge documents issue classes rather than a row-by-row answer key; each correction must be justified from the field definitions.
• —Spend is USD and assumed additive across campaigns.
• —Conversions use the dataset's synthetic last-touch attribution and do not establish causal lift.

Open the matching dataset and exercise.