Use these patterns as starting points. Read the official references alongside each week's activity.

Project 1 — Requirements & Deliverables

What to Deliver

Acceptance Criteria

What to Submit

5.1 ERD Design Checklist

Before finalising your ERD, verify all of these:

5.2 Star Schema vs Snowflake Schema

5.3 SQL Patterns to Know (Week 1)

-- Generic fact table example (adapt column names to your actual dataset)
CREATE TABLE fact_sales (
  sale_id      SERIAL PRIMARY KEY,
  product_id   INT REFERENCES dim_products(product_id),
  customer_id  INT REFERENCES dim_customers(customer_id),
  date_id      INT REFERENCES dim_date(date_id),
  quantity     INT,
  amount       DECIMAL(10,2)
);

-- Star schema query: total sales by category and year
SELECT
  p.category,
  d.year,
  SUM(f.amount)   AS total_revenue,
  COUNT(*)        AS total_orders
FROM fact_sales f
JOIN dim_products  p ON f.product_id  = p.product_id
JOIN dim_date      d ON f.date_id     = d.date_id
WHERE d.year = 2024
GROUP BY p.category, d.year
ORDER BY total_revenue DESC;

-- Window function: rank items by value within a group
SELECT
  p.category,
  p.name,
  SUM(f.amount)  AS revenue,
  RANK() OVER (PARTITION BY p.category ORDER BY SUM(f.amount) DESC) AS rank_in_category
FROM fact_sales f
JOIN dim_products p ON f.product_id = p.product_id
GROUP BY p.category, p.name;

Week 1 Official References

Project 2 — Requirements & Deliverables

What to Build

Acceptance Criteria

What to Submit

5.4 Shell Script Template

#!/bin/bash
set -e   # exit immediately on any error

# Replace with the actual dataset URL for your activity
DATA_URL="https://example.com/dataset.csv"
OUTPUT_DIR="./data/raw"
OUTPUT_FILE="$OUTPUT_DIR/dataset.csv"

mkdir -p "$OUTPUT_DIR"

echo "[$(date)] Starting download from $DATA_URL"
curl -L -o "$OUTPUT_FILE" "$DATA_URL"
ROW_COUNT=$(wc -l < "$OUTPUT_FILE")
echo "[$(date)] Download complete: $ROW_COUNT lines saved to $OUTPUT_FILE"

5.5 PySpark: Load CSV to PostgreSQL

# Adapt schema fields to match your actual dataset
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, FloatType

spark = SparkSession.builder \
    .appName("DataPipeline") \
    .config("spark.jars", "/path/to/postgresql-42.7.3.jar") \
    .getOrCreate()

# Define schema to match your CSV columns
schema = StructType([
    StructField("id",       IntegerType(), True),
    StructField("name",     StringType(),  True),
    StructField("category", StringType(),  True),
    StructField("year",     IntegerType(), True),
    StructField("value",    FloatType(),   True),
])

df = spark.read.csv("./data/raw/dataset.csv", header=True, schema=schema)

DB_URL = "jdbc:postgresql://localhost:5432/bootcamp"
PROPS = {"user":"postgres","password":"postgres","driver":"org.postgresql.Driver"}

df.write.jdbc(url=DB_URL, table="staging_raw", mode="overwrite", properties=PROPS)
print(f"Loaded {df.count()} rows into staging_raw")

5.6 Pandas: Clean and Write

# Adapt column names and cleaning rules to match your actual dataset
import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("postgresql://postgres:postgres@localhost:5432/bootcamp")

# Read from staging table
df = pd.read_sql("SELECT * FROM staging_raw", engine)

# 1. Drop duplicate rows
df = df.drop_duplicates(subset=["id"])

# 2. Fill missing category values
df["category"] = df["category"].fillna("Unknown")

# 3. Remove rows with no name
df = df.dropna(subset=["name"])

# 4. Normalise text columns
df["name"]     = df["name"].str.strip().str.title()
df["category"] = df["category"].str.strip().str.lower()

# 5. Remove rows with invalid numeric values
df = df[df["value"] >= 0]

# Write clean data
df.to_sql("clean_data", engine, if_exists="replace", index=False)
print(f"Wrote {len(df)} clean rows to clean_data")

5.7 SQL Patterns for Analytics

-- Window function: rank records by value within each category
SELECT
  name,
  category,
  value,
  RANK() OVER (PARTITION BY category ORDER BY value DESC) AS category_rank
FROM clean_data;

-- LAG / LEAD: year-over-year change in record counts
SELECT
  year,
  COUNT(*)                                        AS record_count,
  LAG(COUNT(*)) OVER (ORDER BY year)              AS prev_year_count,
  COUNT(*) - LAG(COUNT(*)) OVER (ORDER BY year)  AS yoy_change
FROM clean_data
GROUP BY year
ORDER BY year;

Week 2 Official References

Project 3 — Requirements & Deliverables

What to Build

Acceptance Criteria

What to Submit

5.8 dbt Project Structure

my_project/
  dbt_project.yml
  profiles.yml
  models/
    staging/
      stg_source_a.sql
      stg_source_b.sql
      _stg_models.yml      # sources + tests
    intermediate/
      int_records_enriched.sql
    marts/
      fct_transactions.sql
      dim_categories.sql
      _mart_models.yml
  tests/
    assert_value_range.sql
  macros/

5.9 dbt Model Templates

Staging model — stg_source_a.sql

-- models/staging/stg_source_a.sql
-- Rename and cast columns from the raw source — no business logic here
WITH source AS (
    SELECT * FROM {{ source("raw", "source_a") }}
),

renamed AS (
    SELECT
        id::INT                 AS record_id,
        TRIM(name)              AS name,
        LOWER(TRIM(category))   AS category,
        year::INT               AS year,
        value::FLOAT            AS value
    FROM source
    WHERE id IS NOT NULL
)

SELECT * FROM renamed

Mart model — fct_transactions.sql

-- models/marts/fct_transactions.sql
-- Final mart layer: join staging and enriched intermediate models
WITH records AS (
    SELECT * FROM {{ ref("stg_source_a") }}
),

enriched AS (
    SELECT * FROM {{ ref("int_records_enriched") }}
),

final AS (
    SELECT
        r.record_id,
        r.name,
        r.category,
        r.year,
        e.avg_value,
        e.total_count
    FROM records r
    LEFT JOIN enriched e USING (record_id)
)

SELECT * FROM final

Schema tests — _mart_models.yml

# models/marts/_mart_models.yml
version: 2

models:
  - name: fct_transactions
    description: "One row per record with aggregated metrics."
    columns:
      - name: record_id
        description: "Unique record identifier"
        tests:
          - unique
          - not_null
      - name: category
        description: "Normalised category string"
        tests:
          - not_null
      - name: avg_value
        description: "Mean value across all transactions"

5.10 Airflow DAG Template

# dags/data_pipeline.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator

default_args = {
    "owner": "intern",
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    dag_id="data_pipeline",
    default_args=default_args,
    schedule="@daily",
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    extract = BashOperator(
        task_id="extract",
        bash_command="bash /opt/airflow/scripts/download.sh",
    )

    load = BashOperator(
        task_id="load",
        bash_command="python /opt/airflow/scripts/load_pyspark.py",
    )

    transform = BashOperator(
        task_id="dbt_run",
        bash_command="dbt run --project-dir /opt/airflow/dbt",
    )

    test = BashOperator(
        task_id="dbt_test",
        bash_command="dbt test --project-dir /opt/airflow/dbt",
    )

    extract >> load >> transform >> test

Weeks 3–5 Official References

Project 4 — Requirements & Deliverables

What to Build

Acceptance Criteria

What to Submit

5.11 Power BI Data Model Checklist

5.12 DAX Measure Patterns

-- Adapt table and column names to your actual data model

-- Basic measures
Total Records = COUNTROWS(fct_transactions)
Avg Value = AVERAGE(fct_transactions[avg_value])
Total Count = SUM(fct_transactions[total_count])

-- % of total (within a category)
Category Share % =
DIVIDE(
    [Total Records],
    CALCULATE([Total Records], ALL(dim_categories[name]))
)

-- Conditional measure
Top Performers =
CALCULATE(
    [Total Records],
    fct_transactions[avg_value] >= 100
)

-- Year-over-year comparison (requires date table)
Records Last Year =
CALCULATE([Total Records], SAMEPERIODLASTYEAR(dim_date[date]))

5.13 Data Storytelling Principles

Weeks 6–7 Official References