Transformation Fundamentals

Transformation is the stage where raw, untrustworthy data becomes data people can build decisions on. It is the "T" in ETL/ELT and the part most people picture when they hear "data pipeline". This chapter teaches the fundamentals that apply no matter which tool (SQL, dbt, Spark) you use.

The core idea is that almost every transformation is one of three jobs, cleaning, shaping, or aggregating, applied in that order. Get the jobs and the order right and your transformations are correct and easy to reason about. Get them wrong, usually by aggregating before cleaning, and you produce confident, wrong numbers.

By the end you will be able to take a messy raw table and turn it into a trustworthy metric step by step, and explain why each step happens where it does. You will also understand why set-based thinking (one operation over all rows) beats row-by-row loops at the scales data engineers actually work at.

transformation

Turning raw data into trustworthy data by cleaning, shaping, and aggregating it.

cleaning

Making data valid: dropping bad rows, fixing types, standardising, de-duplicating.

shaping

Combining data: filtering, joining, and enriching into one coherent structure.

aggregating

Summarising rows into metrics: counts, sums, and rollups.

Every transformation, however complex, decomposes into three kinds of work. Cleaning makes data valid: dropping malformed rows, fixing types, standardising formats, removing duplicates. Shaping combines data: filtering to what you need, joining tables together, enriching rows with context from elsewhere. Aggregating summarises: grouping rows into counts, sums, and the rollups a business reads.

The order matters because each job assumes the previous one is done. Shaping assumes the data is valid (you do not want to join on a malformed key). Aggregating assumes the data is both valid and correctly combined (you do not want to sum amounts that include test rows). Run them out of order and errors compound silently.

This decomposition is also how you read and debug someone else’s transformation. Faced with a 200-line SQL model, you ask: where does it clean, where does it shape, where does it aggregate? Naming the three jobs turns an intimidating wall of code into three understandable phases.

validation

Checking and enforcing that data meets expectations (types, ranges, non-null).

join

Combining rows from two tables on a matching key, a core shaping operation.

enrichment

Adding context to rows by joining in reference data (e.g. country for a customer).

rollup

An aggregation that summarises detailed rows into higher-level totals.

Cleaning is the first and most underrated job: making raw data valid before anything else touches it. Raw data is dirty by default, duplicate rows from retries, wrong types (a number stored as text), inconsistent formats ("USA" vs "United States"), nulls where values are required, and test or bot rows that should never count.

The example shows the most common cleaning operations on raw events: drop rows with a null user, remove duplicates by a key, exclude bots, and standardise the country code. Each is a rule, and the power of encoding cleaning as rules is that they run identically every time, unlike a human cleaning a spreadsheet by eye.

A subtle but important decision is what to do with bad rows: drop them, fix them, or quarantine them. Silently dropping data hides problems; a good pipeline often routes invalid rows to a separate "quarantine" table so they can be inspected, rather than vanishing. Cleaning should be visible, not a silent deletion.

de-duplication

Removing repeated rows, usually by a key, often caused by retries upstream.

type casting

Converting a value to its correct type (text "42" to integer 42).

standardisation

Making formats consistent (casing, units, date formats) across rows.

quarantine

Routing invalid rows to a separate table for inspection instead of dropping them silently.

-- Cleaning rules: valid user, de-duplicated, no bots, standard country.
select   distinct on (event_id)            -- de-duplicate by key
         event_id,
         user_id,
         upper(country)        as country,  -- standardise format
         event_time
from     raw_events
where    user_id is not null               -- drop invalid rows
  and    is_bot = false                    -- exclude bots
order by event_id, event_time;

Once data is valid, shaping combines it into the structure you actually need. The three core operations are filtering (keeping only relevant rows), joining (combining tables on a key), and enriching (adding context from reference data). Shaping is where isolated tables become a coherent, analysable picture.

The example enriches cleaned events with the user’s country and plan from a users table, a join. The result is one table that carries everything a later aggregation needs, so you do not have to re-join repeatedly downstream. Good shaping front-loads the joins so the aggregation step is simple.

The danger in shaping is the join itself. Join on a non-unique key and rows multiply (the "fan-out" trap): one event matching three user rows becomes three events, silently tripling your counts. Always know the cardinality of a join, one-to-one or one-to-many, before you write it, and verify the row count after.

filter

Keeping only the rows relevant to the task (a WHERE clause).

join cardinality

How many rows on each side match: one-to-one, one-to-many, or many-to-many.

fan-out

A join on a non-unique key multiplying rows and silently inflating downstream counts.

reference data

Lookup data (countries, plans, product catalog) joined in to enrich rows.

-- Shape: attach each user's plan and country to their events.
-- users.user_id is unique, so this one-to-many join does not fan out.
select   e.event_id,
         e.user_id,
         u.plan,
         u.country
from     cleaned_events e
join     users u on u.user_id = e.user_id;

Aggregation is the final job: collapsing many detailed rows into the summary numbers a business reads. Counts, sums, averages, and distinct counts turn thousands of cleaned, shaped events into "daily active users" or "revenue by country". This is where data finally becomes a decision-ready metric.

The example groups enriched events by country and counts distinct users, producing active users per country. Notice it only works because the data was already cleaned (no bots or duplicates to inflate it) and shaped (country was joined in). Aggregation is the payoff of the two jobs before it.

The defining decision in aggregation is the grain: what does one output row represent? "One row per country per day" is a grain. Mixing grains, or aggregating at the wrong one, produces numbers that cannot be compared or summed correctly. State the grain of your output explicitly, it is the single most important thing about an aggregated table.

aggregate function

A function that summarises rows: COUNT, SUM, AVG, COUNT(DISTINCT).

grain

What one output row represents (e.g. one row per country per day).

group by

The clause that defines the aggregation grain by grouping rows.

double counting

Inflating an aggregate by counting the same underlying entity more than once.

-- Grain: one row per country. Count distinct active users.
-- Works only because the data is already clean and shaped.
select   country,
         count(distinct user_id) as active_users
from     enriched_events
group by country
order by active_users desc;

There are two ways to express a transformation, and at data-engineering scale only one of them works. Row-by-row (procedural) processing loops over each row and handles it individually, the way a beginner naturally thinks. Set-based processing describes the change once and lets the engine apply it to all rows at once.

On a thousand rows, the difference is invisible. On a billion rows, it is the difference between seconds and hours, or never finishing. Set-based operations (a single SQL UPDATE, a Spark transformation) let the engine parallelise across all the data; a row-by-row loop forces it through one item at a time, defeating the very engines built to go fast.

The shift to set-based thinking is one of the biggest mindset changes for engineers coming from general programming. The instinct "loop over the rows and do X to each" must become "describe X as a single operation over the whole set". SQL is set-based by design, which is a large part of why it dominates data transformation.

set-based processing

Expressing a transformation as one operation over an entire dataset.

row-by-row / procedural

Looping over rows and processing each individually; slow at scale.

vectorised / parallel execution

The engine applying an operation across many rows at once.

declarative

Describing the desired result (SQL) rather than the step-by-step procedure.

Clean, shape and aggregate are not just an order — they are three different places in a pipeline. Chapter 2, The Data Pipeline Lifecycle, named those places bronze, silver and gold; this topic walks one day of orders through them with the code that runs at each.

Then two tables: every transformation you will meet and the layer it belongs to, and every tool that can run one, with what it costs.

the layer contract

What a layer is forbidden to do: bronze may not change what arrived, silver may not throw away rows, gold may not be the only copy of anything.

materialisation

Whether a transformation is stored as a table, rebuilt each run, or left as a view. It decides who pays the compute — the pipeline once, or every reader.

One day of Marlow's orders41,206 records in. Walk the three layers and watch what each one is allowed to change.41,206 typed rowsReset

Bronze · stagingSilver · cleanedGold · serving

marlow-bronzeAS RECEIVEDTyped. Same row count.

PySparkgrain — one row per record as received

Land it exactly as it arrived, then make it readable without changing what it says.

in	out
41,206 raw JSON records	41,206 typed rows

• · Cast text to real types
• · Parse timestamps into UTC
• · Add ingested_at and the source file name
• · Quarantine records that will not parse

What runs here · PySpark

# bronze: readable, not corrected. Row count must not change.
raw = (spark.read
       .schema(orders_raw_schema)
       .option("badRecordsPath", "s3://marlow-quarantine/orders/")
       .json("s3://marlow-drop/orders/dt=2026-08-07/"))

bronze = (raw
    .withColumn("total",      col("total").cast("decimal(12,2)"))
    .withColumn("ordered_at", to_utc_timestamp(col("ordered_at"), "Europe/London"))
    .withColumn("ingested_at", current_timestamp())
    .withColumn("source_file", input_file_name()))

bronze.write.mode("overwrite").partitionBy("dt").parquet("s3://marlow-bronze/orders/")

Do this work at the wrong layer and…Deduplicate here and you have destroyed the evidence. Bronze is the copy you re-run from when a silver rule turns out to be wrong — the moment it stops matching what arrived, it cannot do that job.

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, col, lower

latest = Window.partitionBy("order_id").orderBy(col("ingested_at").desc())

silver = (bronze
    .withColumn("rn", row_number().over(latest))
    .filter((col("rn") == 1) & (col("total") >= 0))
    .drop("rn")
    .withColumn("status", lower(col("status")))
    .join(customers, on="customer_id", how="left")     # left: guest checkout survives
)

silver.write.mode("overwrite").partitionBy("dt").parquet("s3://marlow-silver/orders/")

-- models/silver/orders.sql
{{ config(materialized='incremental', unique_key='order_id',
          partition_by={'field': 'dt', 'data_type': 'date'}) }}

SELECT order_id, customer_id, LOWER(status) AS status, total_gbp, ordered_at, dt
FROM {{ ref('bronze_orders') }}
QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY ingested_at DESC) = 1

{% if is_incremental() %}
  WHERE dt >= (SELECT MAX(dt) FROM {{ this }})
{% endif %}

import pandas as pd

df = pd.read_parquet("s3://marlow-bronze/orders/dt=2026-08-07/")

silver = (df
    .sort_values("ingested_at")
    .drop_duplicates("order_id", keep="last")
    .query("total >= 0")
    .assign(status=lambda d: d["status"].str.lower())
    .merge(customers, on="customer_id", how="left"))

silver.to_parquet("s3://marlow-silver/orders/dt=2026-08-07/")

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

val latest = Window.partitionBy($"order_id").orderBy($"ingested_at".desc)

val silver = bronze
  .withColumn("rn", row_number().over(latest))
  .filter($"rn" === 1 && $"total" >= 0)
  .drop("rn")
  .join(customers, Seq("customer_id"), "left")

silver.write.mode("overwrite").partitionBy("dt").parquet("s3://marlow-silver/orders/")

Five scenarios where the transform is where the bug lives — and in four of them nothing anywhere raises an error.

Build them on the pipeline canvas with the chapter closed. Each is graded against the scenario’s own requirement rules rather than against a model answer, so there is more than one design that passes — and a design that does not pass is told exactly which requirement it missed.