The DataFrame API

7 h
requires SPK-02

From your world. dbt staging models, written in PySpark

Validate and deduplicate customers (MDM key)

The customers table covers three companies. The acquired company (C003) reuses the customer_id integers of the parent company. So customer_id = 1 means a different person in each company. Some rows also have a NULL email. Your job is a clean in the style of a dbt staging model:

Implement clean_customers(df) in solution.py. You get the raw customers DataFrame. Return a DataFrame that does these things:

Drops rows with a NULL email (unverified contacts).
Removes duplicates on the business key (company_id, customer_id). It keeps genuinely different people, even when they share a customer_id across companies.
Returns exactly the columns company_id, customer_id, email.

Why the key matters: if you dedup on customer_id alone, you collapse (C001, 1), (C002, 1) and (C003, 1) into one row. That merges three different customers into one. This is the master-data bug that this exercise teaches.

The test compares your result against an independent DuckDB computation on the same data. The test ignores row order and column order.

Starter code

solution.py

"""SPK-03 / 01-clean-customers — starter. See README.md."""


def clean_customers(df):
    # TODO: drop NULL emails, dedup on (company_id, customer_id), select the 3 columns.
    raise NotImplementedError

Hints

Hint 1

Drop rows whose email is NULL: F.col('email').isNotNull(). Remember NULL comparisons are three-valued, so you must test isNotNull explicitly.

Hint 2

The business key is (company_id, customer_id). customer_id alone COLLIDES across companies (the acquired company reuses ids), so deduping on customer_id merges different people — an MDM bug.

Hint 3

Use dropDuplicates(['company_id','customer_id']) and return exactly company_id, customer_id, email. Row/column order are not checked.

Run it locally

uv run de start SPK-03/01-clean-customers
# edit workspace/SPK-03/01-clean-customers/
uv run de test