DSC3103 · Revision guide

Data Science: Systems & Technologies on a few pages

The night-before sheet: every module's one-minute summary, traps, terms, formulas and the answer examiners want.

The 3-beat answer

Examiners reward structure. Every answer, about 30 seconds:

1 · ClaimOne sentence: what it is or does. State the rule directly.

2 · MechanismHow it works: the maths, the steps, the data structures.

3 · Trade-offWhen it fails, what it costs, or why the alternative breaks.

Formulas to know cold

Null rate: null_pct = (missing_count / total_rows) × 100%
IQR & Tukey fences: IQR = Q3 − Q1, bounds = [Q1 − 1.5·IQR, Q3 + 1.5·IQR]
Row conservation (sample_qtn): rows_in = rows_kept + rows_rejected + exact_duplicates_removed
Idempotency law: f(f(D)) = f(D) (state after N runs = state after 1 run)
Columnar scan savings: IO_parquet ≈ IO_row × (k_selected / k_total) ÷ compression_ratio
Star schema fact grain: |Fact| ≤ |dim_date| × |dim_market| × |dim_commodity|
Unknown dimension mapping: farmer_key = COALESCE(dim_farmer.farmer_key, 0) -- 0 = 'Unknown'
Window top-N per group: RANK() OVER (PARTITION BY centre ORDER BY total_litres DESC) <= 3
Airflow exponential backoff: delay(k) = base_delay × 2^(k−1), total = base × (2^R − 1)
Spark partition waves: waves = ⌈num_partitions / total_executor_cores⌉
Partial aggregate reduction: rows_shuffled ≤ num_partitions × num_distinct_keys
Tumbling window & watermark: W_start = ⌊ts/S⌋·S, Watermark W(t) = max(event_ts) − L
Serving batch throughput: T_N = T_startup_load + N · T_infer vs N · (T_load + T_infer)

Module 1

Data Profiling, Validation & Cleaning

Raw data is untrusted and immutable: never edit data/raw/, always write deterministic outputs to data/processed/. Profile first (types, null rates, duplicates, quantiles), then validate the data contract, then clean: standardise strings, drop rows missing the composite key, deduplicate on (date, market, commodity), and impute prices with the per-commodity median because prices are skewed and differ by commodity.

Imputing with the global mean — outliers and different commodities distort it.
Overwriting the raw CSV so a cleaning bug can never be undone.
Deduplicating on all columns instead of the business key.

Data profiling: Inspecting row counts, types, null rates, uniqueness and distributions before cleaning.
Data contract: Formal schema + domain rules (e.g. price_ugx > 0, rainfall_mm in [0, 500]) that data must pass.
Composite business key: Columns that uniquely identify an observation: (date, market, commodity).
IQR outlier rule: Flag values outside [Q1 − 1.5·IQR, Q3 + 1.5·IQR], where IQR = Q3 − Q1.
Immutability of raw: Never mutate data/raw/; always write deterministic outputs to data/processed/.

Likely question: Why should missing market prices be imputed using the per-commodity median rather than the overall dataset mean, and why is raw data kept strictly immutable?

Market prices are skewed, contain extreme outliers, and vary across commodities (e.g., beans vs maize). Using a global mean mixes incompatible distributions and is sensitive to outliers, whereas the per-commodity median is robust and preserves each commodity's typical price level. Raw data must remain immutable so the pipeline is reproducible and auditable: if cleaning logic changes or has a bug, we can re-run from the untouched source CSV without permanent data loss.

Module 2

Pipelines: ETL vs ELT, Parquet & Idempotency

A pipeline is modular stages (profile → validate → clean → store) that can be re-run safely. ETL transforms before loading; ELT loads raw first and transforms in-engine. Parquet is columnar, typed and compressed, so queries read only the columns they need. Idempotency means running twice leaves the same state: use atomic temp-file swaps (os.replace) and primary-key upserts or partition overwrites, never blind appends.

Using append mode — every retry or backfill duplicates rows.
Writing directly to the final file so a crash leaves it half-written.
Thinking Parquet compression is the only win — column pruning is usually bigger.

Idempotency: Running a job N times has the exact same effect as running it once.
Columnar storage (Parquet): Stores values column-by-column; enables column pruning and high compression ratios.
Row group & predicate pushdown: Parquet stores min/max statistics per block so engines skip irrelevant chunks without reading them.
Atomic write: Write to .tmp then os.replace() so readers never see a truncated or half-written file.
Upsert: INSERT ... ON CONFLICT UPDATE (or INSERT OR REPLACE) keyed on the business key.

Likely question: Define idempotency in data pipelines. Why does a naive append mode break idempotency on retries, and how do you engineer both file outputs and database loads to be idempotent?

A pipeline is idempotent if running it multiple times on the same input leaves the target storage in the exact same state as running it once. Naive append breaks this because any retry or backfill inserts duplicate rows, inflating counts and corrupting aggregates. For file outputs, ensure idempotency by writing to a temporary file and atomically swapping it into place (`os.replace`) or overwriting the specific date partition. For database tables, enforce a composite primary key and use `INSERT OR REPLACE` (upsert) or `DELETE WHERE date = ...` inside a single transaction before inserting.

Module 3

Analytical Star Schemas & DuckDB

OLTP databases handle row-by-row writes; OLAP engines like DuckDB scan few columns over many rows. A star schema has a central fact table of numeric measures and foreign keys at a declared grain, surrounded by descriptive dimension tables used for filtering and grouping. Declare the grain first: mixing grains makes joins fan out and double-count.

Storing descriptive strings (like category names) repeatedly in the fact table.
Mixing daily and monthly rows in one fact table.
Joining a fact to a many-row dimension and silently multiplying SUMs (fan trap).

Grain: The exact business definition of a single row in a fact table (e.g., 1 market × 1 commodity × 1 day).
Fact table: Central table holding numeric measures and foreign keys to dimensions.
Dimension table: Lookup table holding descriptive attributes (who, what, where, when) with a surrogate primary key.
Surrogate key: System-generated key (integer or MD5 hash) decoupling warehouse joins from operational strings.
Vectorised execution: Processing columns in CPU cache-sized batches (vectors of 1,024–2,048 values) instead of row-at-a-time.

Likely question: Contrast Fact tables and Dimension tables in a Star Schema. Why is declaring the exact 'grain' of a fact table critical before designing joins and aggregations?

Fact tables store quantitative numeric measurements (such as `price_ugx` and `rainfall_mm`) and foreign keys at a specific atomic event level, making them narrow and deep. Dimension tables (`dim_market`, `dim_commodity`, `dim_date`) store descriptive attributes used for `WHERE` filtering and `GROUP BY` labels. Declaring the grain — exactly what one fact row represents, such as one commodity in one market on one day — is critical because mixing grains causes joins to fan out, silently double-counting sums and biasing averages.

Module 4

Workflow Orchestration with Apache Airflow

Airflow models a pipeline as a DAG of tasks with explicit dependencies, so a failed upstream task marks downstream tasks upstream_failed and bad data never reaches storage. Give network-facing ingestion several retries with exponential backoff; let deterministic validation fail fast. Because of retries and backfills, every task must be idempotent.

Forgetting a dependency edge (e.g. stored >> report_gen) and creating a race condition.
Retrying deterministic schema failures — they won't fix themselves.
Passing whole DataFrames through XCom instead of small metadata like paths.

DAG: Directed Acyclic Graph: tasks with directed dependencies and zero loops.
Operator / TaskFlow: Template defining a unit of work (`@task` Python decorator or `BashOperator`).
XCom: Cross-communication mechanism for passing small metadata (like file paths or row counts) between tasks.
Backfill & Catchup: Running a DAG across historical logical date intervals (`data_interval_start`).
Exponential backoff: Doubling the wait time between consecutive retry attempts: d, 2d, 4d, …

Likely question: In an Airflow DAG (`ingest >> validate >> clean >> store >> report`), why should retry counts and delays differ by stage, and what happens if `validate` fails?

Ingestion interacts with external networks, APIs, or remote filesystems prone to transient timeouts, so it benefits from higher retries (e.g., 2–3 retries with backoff). Validation and cleaning run deterministic logic on local data; if a schema check fails, retrying immediately wastes compute without fixing the broken contract. When `validate` fails, Airflow marks it `failed` and sets all downstream tasks (`clean`, `store`, `report`) to `upstream_failed`, halting the pipeline so corrupt data never overwrites production tables.

Module 5

Distributed Batch Processing with PySpark

Spark splits DataFrames into partitions processed in parallel by executors; one partition is one task, and stages run in waves of ceil(partitions / cores). Transformations are lazy until an action runs. Narrow transformations stay inside a partition; wide ones (groupBy, join) shuffle data across the network. On small data JVM and shuffle overhead dominate, so DuckDB can win and .cache() can even slow things down.

Expecting a transformation to run before an action is called.
Caching a DataFrame that is only used once.
Choosing Spark for data that fits comfortably on one machine.

Lazy evaluation: Transformations build a query plan; execution only starts when an Action is called.
Narrow transformation: 1-to-1 partition mapping (`filter`, `select`); pipelined in memory with no shuffle.
Wide transformation: Many-to-many partition mapping (`groupBy`, `join`); inserts an `Exchange` shuffle barrier.
Catalyst & Tungsten: Spark's query optimiser (predicate pushdown, column pruning) and off-heap code generation engine.
Partial aggregate: Map-side `HashAggregate` that pre-combines groups inside each partition before shuffling.

Likely question: Explain the difference between narrow and wide transformations in PySpark. Why did DuckDB beat PySpark on our 10,000-row market dataset, and when does `.cache()` actually pay off?

Narrow transformations (`filter`, `select`, `withColumn`) compute each output partition from a single input partition in memory without moving data. Wide transformations (`groupBy`, `join`, `distinct`) require a shuffle (`Exchange`), redistributing rows across partitions via disk and network I/O. On a 10,000-row dataset, DuckDB (~1.4s) beats PySpark (~11.2s) because Spark pays heavy fixed overhead for JVM startup, Catalyst planning, task scheduling, and shuffle serialisation. Similarly, `.cache()` has an upfront materialisation cost and only pays off when a large, expensive DataFrame is reused across multiple downstream actions.

Module 6

Streaming, Event-Time Windows & Watermarks

Streaming processes unbounded events continuously. Always bucket by event time, not processing time, because mobile networks deliver events late and out of order. Tumbling windows are [floor(ts/S)·S, +S). The watermark W(t) = max_event_ts − L says when a window can be finalised; events older than the watermark go to late_events, and a seen_ids store drops duplicate retries.

Grouping by processing time so delayed events land in the wrong window.
Setting allowed lateness to 0 and losing every slightly late event.
Deduplicating after windowing instead of before.

Event Time vs Processing Time: When the event happened in the real world vs when the server processed it.
Tumbling window: Fixed-size, non-overlapping time intervals: `[floor(t/W)·W, floor(t/W)·W + W)`.
Watermark: Moving threshold `max_event_time − L` signalling that windows ending at or before `W` are complete.
Allowed lateness (L): Tolerance window for out-of-order events; trades off completeness against output latency.
Stateful deduplication: Tracking seen `event_id` keys so producer network retries do not double-count metrics.

Likely question: Distinguish Event Time from Processing Time. How does a Watermark control when a tumbling window is finalised, and what happens to an event whose timestamp falls behind the watermark?

Event time is when the event actually occurred at the source, whereas processing time is when the consumer ingests it. Because network delays cause events to arrive out of order, we group by event-time windows and track a watermark `W(t) = max(event_time) − L`, where `L` is the allowed lateness. Once `W(t) >= window_end`, the engine finalises and emits that window's aggregate. Any event arriving with `event_time < W(t)` is classified as late and routed to a side output (`late_events.jsonl`) or dead-letter stream rather than silently corrupting closed windows.

Module 7

From Model to Data Product: FastAPI & Docker

A model becomes a data product when it sits behind a validated HTTP API. Save the model with its metadata (version, feature columns, metrics), load it once at startup rather than per request, validate inputs with strict Pydantic schemas (HTTP 422 for bad payloads), and share feature code between training and serving to prevent training-serving skew. Package it with pinned dependencies in Docker.

Loading model.joblib inside the /predict handler.
Re-implementing feature logic in the API (training-serving skew).
Unpinned requirements so the container differs from training.

Training-serving skew: Mismatch between feature engineering or data distribution at training time vs inference time.
Pydantic schema validation: Declarative type + range checks (`Literal`, `Field(ge=0, le=500)`) that reject bad payloads with HTTP 422.
Startup model loading: Deserialising `model.joblib` once into memory when the worker starts rather than per HTTP request.
Model metadata artifact: `model_metadata.json` storing `model_version`, `trained_at`, `feature_columns`, and test metrics (`r2`).
Docker layer caching: Copying `requirements.txt` and running `pip install` before `COPY src/` so code edits rebuild in seconds.

Likely question: What is training-serving skew in ML systems? Explain how a subtle threshold mismatch causes silent prediction errors, and how shared feature modules, Pydantic schemas, and Docker work together to make a model service reproducible.

Training-serving skew occurs when the feature transformations applied during live inference differ from those used during model training. Because the model still receives a numeric vector of the expected shape, it raises no HTTP or runtime error — instead, it silently outputs wrong predictions (such as when `is_rainy` threshold drifted from 5.0 mm to 10.0 mm in our lab). We eliminate this by defining `make_features()` and `FEATURE_COLUMNS` in a single shared module (`src/common/features.py`) imported by both `train.py` and `main.py`, validating raw request bounds with Pydantic (returning HTTP 422 on bad inputs), loading `model.joblib` once at startup, and locking the OS and Python dependencies in a Dockerfile.

Module 8

Practice Exam (sample_qtn): Milk Validation & Collisions

In sample_qtn, tidy formatting (strip().title() on centre) before running membership rules, or valid deliveries get flagged as unknown centres. Exact duplicates are double-saved records — keep one. A record_id collision has the same id but different values, so both rows are quarantined in rejected_df. Always assert row conservation: rows_in = kept + rejected + exact_duplicates_removed.

Running rule_unknown_centre before tidying centre text.
Keeping the first row of a record_id collision.
Not asserting the row-conservation invariant at the end.

Exact duplicate: All columns identical (`df.duplicated(keep='first')`); drop extra copies and keep 1 row.
record_id collision: Same `record_id` but different column values (`duplicated(subset=['record_id'], keep=False)` after dropping exact dupes); reject ALL colliding rows.
Row conservation invariant: `rows_in == rows_kept + rows_rejected + exact_duplicates_removed`.
Boundary checksum: SHA-256 hash of `data/raw/milk_raw.csv` logged at pipeline start to prove raw input integrity.
Lossless reference join: `LEFT JOIN` (or validated inner join after rejecting unknown centres) asserting `len(joined) == len(clean_df)`.

Likely question: In `sample_qtn.pdf`, why must validation rules on `centre` run AFTER tidying text, why are BOTH rows of a `record_id` collision rejected instead of keeping one, and what makes running the pipeline twice safe?

Validation rules on `centre` must run AFTER stripping whitespace and standardising case (`strip().title()`) so valid centres like `' mbarara '` are not falsely rejected as unknown. For a `record_id` collision (same ID, different payload values), we reject BOTH rows because there is no way to know which delivery is genuine — keeping one arbitrarily would corrupt financial payouts, whereas exact duplicates have identical values so keeping one is safe. Running the pipeline twice is safe because `data/raw/milk_raw.csv` and `centres.csv` are never modified, cleaning is deterministic, and outputs overwrite `data/processed/q1_output.parquet` and DuckDB tables atomically.

Module 9

Practical Test 02 (dsc3103_test_02): Milk Star Schema & Analytics

In test_02, 6 farmers (408 deliveries) are missing from farmers.csv, so add an Unknown member (farmer_key = 0) and COALESCE the join key — all 20,000 fact rows survive with no NULL keys. capacity_litres is a DAILY limit: aggregate litres per (centre, date) in a CTE before comparing. Rank top farmers with RANK() OVER (PARTITION BY centre …) excluding Unknown, and partition by month so April queries scan one partition.

Inner-joining dim_farmer and silently dropping 408 deliveries.
Comparing single deliveries to a daily capacity.
Including the Unknown farmer bucket in top-farmer rankings.

Unknown member row (`key = 0`): Default dimension record (`'Unknown'`) so unregistered business keys map cleanly without NULL FKs or dropped facts.
Pre-computed measure (`revenue_ugx`): Stored once in `fact_delivery` (`litres * price_ugx_per_litre`) for consistent, fast aggregations.
Two-stage CTE aggregation: Aggregate atomic deliveries to daily centre totals in a CTE before comparing against daily `capacity_litres`.
Window ranking (`RANK() OVER`): `RANK() OVER (PARTITION BY centre ORDER BY total_litres DESC)` to pick top-N per group.
SCD Type 2: Slowly Changing Dimension technique that adds a new versioned dimension row when an attribute changes, preserving history.

Likely question: In `dsc3103_test_02`, how should `dim_farmer` and `fact_delivery` handle the 6 unregistered `farmer_id`s so all 3 integrity checks pass? Also explain why `days_over_capacity` needs a two-stage aggregation and what happens when a farmer moves centres in July.

To preserve all 20,000 deliveries without NULL foreign keys, `dim_farmer` includes a dedicated Unknown member row (`farmer_key = 0, farmer_id = -1, farmer_name = 'Unknown'`) and `fact_delivery` maps unmatched `farmer_id`s via `COALESCE(f.farmer_key, 0)`. That satisfies all three integrity checks: `COUNT(*) == 20,000`, zero orphan foreign keys, and zero NULL keys. For `days_over_capacity`, `capacity_litres` is a daily centre limit, so a CTE must first compute `SUM(litres)` per `(centre, date_key)` before counting days where `daily_litres > capacity_litres`. When a farmer moves centres in July, earlier deliveries in `fact_delivery` keep their original `centre_key` so history is untouched; to also track home-centre history in `dim_farmer`, use SCD Type 2 (a new dimension row with `valid_from`/`valid_to`).