Solutions — Homework 13

Lesson 13 · Data Engineering with Python · Schema Evolution & Lineage

Task 1 — Migration plan for an unsafe change

The worked example below uses the rename actor → actor_username case. The structure of a correct plan is the same for the other two options (remove payload_size, narrow id).

Step 1: Dual-write window (minimum 30 days)

The data_contract.yml deprecation_policy requires removal_notice_minimum_days: 30. During the dual-write window the producer — the transform step in gharchive_hourly.py — must emit both columns in every output row: actor populated with the original value, and actor_username populated with the same value. Neither column is empty; both are present in every Parquet file and every warehouse table written during the transition period. The v4 data contract is bumped to v5 with both fields declared:

  - { name: actor,          type: string, required: true,  added_in: v1,
      deprecated_in: v5,    removal_target: v6 }
  - { name: actor_username, type: string, required: true,  added_in: v5 }

The dual-write window begins on the day the producer deploys the v5 schema. Removal cannot happen before 30 calendar days have elapsed and all consumers have migrated.

Step 2: Consumer audit

You need to enumerate every downstream system that reads the actor column by name. In the sample pipeline the consumers are:

• dbt models — grep the dbt/models/ directory for actor references: grep -rn "actor" dbt/models/. Each hit is a consumer to update.
• Ad-hoc notebooks — search all .ipynb files in the repository for "actor" column selections.
• Downstream dashboards — any BI tool (Metabase, Superset, Tableau) that queries dim_events.actor directly. These are the hardest to find; maintain a registry or ask every dashboard owner to confirm.
• OpenLineage lineage graph — query Marquez for all datasets that have actor in their column-level facets to discover consumers you may not know about.

Track progress in a checklist. Removal is blocked until every known consumer has been updated and the checklist is complete.

Step 3: Removal

After at least 30 days AND zero remaining references to actor in the consumer checklist, drop the column. Update the data contract to v6:

version: v6
fields:
  - { name: actor,          type: string, required: true,  added_in: v1,
      deprecated_in: v5,    removed_in: v6 }
  - { name: actor_username, type: string, required: true,  added_in: v5 }

The removed_in: v6 annotation preserves history: anyone reading the contract can see that the field existed, when it was deprecated, and when it was removed. A reader of v5 Parquet files can still find actor in those files; a reader of v6 files will not.

Task 2 — dbt model contract

The models.yml file

version: 2
models:
  - name: dim_events
    config:
      contract:
        enforced: true
    columns:
      - { name: id,         data_type: bigint,     constraints: [{type: not_null}] }
      - { name: actor,      data_type: varchar }
      - { name: created_at, data_type: timestamp }

Expected dbt build output (clean run)

Running with dbt=1.7.x
...
1 of 1 START sql table model main.dim_events ..................... [RUN]
1 of 1 OK created sql table model main.dim_events ............... [OK in 0.12s]
Finished running 1 table model in 0.xx seconds.

Completed successfully.
All checks passed!

Expected dbt build output (type mismatch deliberately introduced)

Change id’s data_type in the contract to varchar while the model SQL still casts id as bigint. The build fails with:

Database Error in model dim_events
  Contract enforcement failed for model dim_events:
  - Column 'id': data type mismatch
    Expected:  varchar
    Got:       bigint

  Compilation Error
  Contract violations found for model dim_events

The error message names the column, the expected type (from the contract YAML), and the actual type (from the compiled SQL). This is exactly the enforcement you want: a PR that changes either the contract or the model in an incompatible way fails CI before it reaches production.

Task 3 — Trace lineage one hop upstream

Worked example using the file backend

After completing the lab (Steps 1–7), at least two event files exist under observability/events/: a START and a COMPLETE event for the evolve_events_v4 job. Inspecting the COMPLETE event:

python -c "
import json, pathlib
p = pathlib.Path('data-engineering/sample-pipeline/observability/events/')
files = sorted(p.glob('*-COMPLETE.json'), key=lambda f: f.stat().st_mtime)
if files:
    print(json.dumps(json.loads(files[-1].read_text()), indent=2))
"

The output will include fields like:

{
  "eventType": "COMPLETE",
  "eventTime": "2026-06-03T14:22:05.123456Z",
  "run": {
    "runId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
  },
  "job": {
    "namespace": "default",
    "name": "evolve_events_v4"
  },
  "inputs": [
    { "namespace": "default", "name": "events_v3" }
  ],
  "outputs": [
    { "namespace": "default", "name": "events_v4" }
  ]
}

From this event you can state: the most recent upstream job that produced events_v4 was named evolve_events_v4, with run-id a1b2c3d4-..., completing at 2026-06-03T14:22:05Z. Its input was events_v3.

Worked example using Marquez

In the Marquez UI (http://localhost:3000):

1. Click Datasets in the left panel. Find events_v4.
2. Click the dataset, then click Lineage. You see an upstream node labeled evolve_events_v4 connected by a directed edge to events_v4.
3. Click the job node. The Runs tab shows a timeline with one completed run. Expand the run to read the runId and completion timestamp.
4. Click the upstream dataset node (events_v3) to confirm it is the input.

The Marquez column-lineage view (if you have column-level facets enabled) also shows that the country column in events_v4 originated from the evolve_events_v4 transform, not from events_v3 (which has no country column).