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)

-- Create a fact table with foreign keys
CREATE TABLE fact_movie_ratings (
  rating_id   SERIAL PRIMARY KEY,
  movie_id    INT REFERENCES dim_movies(movie_id),
  user_id     INT REFERENCES dim_users(user_id),
  date_id     INT REFERENCES dim_date(date_id),
  rating      DECIMAL(3,1),
  review_len  INT
);

-- Basic star schema query
SELECT
  m.title,
  m.genre,
  AVG(f.rating)  AS avg_rating,
  COUNT(*)       AS total_reviews
FROM fact_movie_ratings f
JOIN dim_movies m ON f.movie_id = m.movie_id
JOIN dim_date  d ON f.date_id  = d.date_id
WHERE d.year = 2024
GROUP BY m.title, m.genre
ORDER BY avg_rating DESC;

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

DATA_URL="https://example.com/movies.csv"
OUTPUT_DIR="./data/raw"
OUTPUT_FILE="$OUTPUT_DIR/movies.csv"

mkdir -p "$OUTPUT_DIR"

echo "[$(date)] Starting download from $DATA_URL"
curl -L -o "$OUTPUT_FILE" "$DATA_URL"
echo "[$(date)] Downloaded $(wc -l < $OUTPUT_FILE) lines to $OUTPUT_FILE"

5.5 PySpark: Load CSV to PostgreSQL

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, FloatType

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

schema = StructType([
    StructField("movie_id",    IntegerType(), True),
    StructField("title",       StringType(),  True),
    StructField("genre",       StringType(),  True),
    StructField("release_year",IntegerType(), True),
    StructField("rating",      FloatType(),   True),
])

df = spark.read.csv("./data/raw/movies.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_movies", mode="overwrite", properties=PROPS)
print(f"Loaded {df.count()} rows into staging_movies")

5.6 Pandas: Clean and Write

import pandas as pd
from sqlalchemy import create_engine

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

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

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

# 2. Fill missing genres
df["genre"] = df["genre"].fillna("Unknown")

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

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

# 5. Clamp ratings to valid range
df = df[df["rating"].between(0, 10)]

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

5.7 SQL Patterns for Analytics

-- Window function: rank movies by rating within each genre
SELECT
  title,
  genre,
  rating,
  RANK() OVER (PARTITION BY genre ORDER BY rating DESC) AS genre_rank
FROM clean_movies;

-- LAG / LEAD: year-over-year change
SELECT
  release_year,
  COUNT(*)                                            AS movie_count,
  LAG(COUNT(*)) OVER (ORDER BY release_year)         AS prev_year_count,
  COUNT(*) - LAG(COUNT(*)) OVER (ORDER BY release_year) AS yoy_change
FROM clean_movies
GROUP BY release_year
ORDER BY release_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_movies.sql
      stg_ratings.sql
      _stg_models.yml      # sources + tests
    intermediate/
      int_movies_enriched.sql
    marts/
      fct_ratings.sql
      dim_movies.sql
      _mart_models.yml
  tests/
    assert_rating_range.sql
  macros/

5.9 dbt Model Templates

Staging model — stg_movies.sql

-- models/staging/stg_movies.sql
WITH source AS (
    SELECT * FROM {{ source("raw", "movies") }}
),

renamed AS (
    SELECT
        movie_id::INT           AS movie_id,
        TRIM(title)             AS title,
        LOWER(TRIM(genre))      AS genre,
        release_year::INT       AS release_year,
        rating::FLOAT           AS rating
    FROM source
    WHERE movie_id IS NOT NULL
)

SELECT * FROM renamed

Mart model — fct_ratings.sql

-- models/marts/fct_ratings.sql
WITH movies AS (
    SELECT * FROM {{ ref("stg_movies") }}
),

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

final AS (
    SELECT
        m.movie_id,
        m.title,
        m.genre,
        m.release_year,
        e.avg_rating,
        e.total_reviews
    FROM movies m
    LEFT JOIN enriched e USING (movie_id)
)

SELECT * FROM final

Schema tests — _mart_models.yml

# models/marts/_mart_models.yml
version: 2

models:
  - name: fct_ratings
    description: "One row per movie with aggregated rating stats."
    columns:
      - name: movie_id
        description: "Unique movie identifier"
        tests:
          - unique
          - not_null
      - name: genre
        description: "Normalised genre string"
        tests:
          - not_null
      - name: avg_rating
        description: "Mean rating across all reviews"

5.10 Airflow DAG Template

# dags/movies_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="movies_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

-- Basic measures
Total Movies = COUNTROWS(fct_ratings)
Avg Rating = AVERAGE(fct_ratings[avg_rating])
Total Reviews = SUM(fct_ratings[total_reviews])

-- % of total
Rating Share % =
DIVIDE(
    [Total Movies],
    CALCULATE([Total Movies], ALL(dim_movies[genre]))
)

-- Conditional measure
High Rated Movies =
CALCULATE(
    [Total Movies],
    fct_ratings[avg_rating] >= 7.5
)

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

5.13 Data Storytelling Principles

Weeks 6–7 Official References