Lab 13 — Evolve the Events Schema from v3 to v4 Safely

Lesson 13 · Data Engineering with Python · ~40 min

By the end of this lab you will have a v4 Parquet fixture with a new nullable country column, the full pytest compatibility suite passing at 9 tests, data_contract.yml bumped to v4, and a lineage event emitted (either to Marquez or to the file backend) recording the schema evolution as a pipeline job. All the code you will extend already exists in data-engineering/sample-pipeline/schema_evolution/ — this lab is a guided tour of adding one forward-compatible column the right way.

Prerequisites

• Lesson 12 lab completed. The sample-pipeline virtual environment must be active with pyarrow and pytest installed.
• If you have not set up the venv yet:

  cd data-engineering/sample-pipeline
  python -m venv .venv && source .venv/bin/activate
  pip install -r requirements.txt

• The schema_evolution/fixtures/ directory must contain at least v1.parquet, v2.parquet, and v3.parquet. If it is empty, regenerate them:

  cd data-engineering/sample-pipeline
  python -m schema_evolution.generate_fixtures

Step 1 — Run the existing pytest suite

Before making any changes, verify the baseline is clean. Run from inside data-engineering/sample-pipeline/:

cd data-engineering/sample-pipeline
pytest schema_evolution/ -v

Expected output (7 tests all passing):

test_v1_columns PASSED
test_v2_adds_optional_payload_size PASSED
test_v3_adds_required_event_type PASSED
test_v3_reads_v1_consumer_safe PASSED
test_v1_reads_v3_producer_safe PASSED
test_new_columns_have_values[v2.parquet-payload_size] PASSED
test_new_columns_have_values[v3.parquet-event_type] PASSED

7 passed

If you see import errors, make sure the venv is active and pyarrow is installed (pip install pyarrow).

Step 2 — Decide on the v4 change

The proposed addition is a nullable country string column, representing the geolocated country of the actor’s profile. This change is backward-compatible:

• It is a new, optional column — existing consumers that select only v1 columns still work.
• It has no NOT NULL constraint — rows without geolocation data carry null rather than breaking.
• It does not rename or remove any existing column.

Step 3 — Generate the v4 fixture

Run the following one-liner from inside data-engineering/sample-pipeline/. It reads the v3 fixture and appends the new column:

python -c "
import pyarrow as pa, pyarrow.parquet as pq
from pathlib import Path

v3 = pq.read_table('schema_evolution/fixtures/v3.parquet')
v4 = v3.append_column('country', pa.array(['US'] * 10, type=pa.string()))
pq.write_table(v4, 'schema_evolution/fixtures/v4.parquet')
print('v4 fixture written:', v4.schema)
"

Verify the file was created:

ls -lh schema_evolution/fixtures/

You should now see v1.parquet, v2.parquet, v3.parquet, and v4.parquet.

Step 4 — Update data_contract.yml

Open data-engineering/sample-pipeline/schema_evolution/data_contract.yml. Make two edits:

1. Bump the top-level version field from v3 to v4.
2. Append the country field entry at the end of the fields list.

The updated file should look like this:

version: v4
namespace: gharchive-pipeline
dataset: events
fields:
  - { name: id,           type: int64,     required: true,  added_in: v1 }
  - { name: actor,        type: string,    required: true,  added_in: v1 }
  - { name: repo,         type: string,    required: true,  added_in: v1 }
  - { name: created_at,   type: timestamp, required: true,  added_in: v1 }
  - { name: payload_size, type: int64,     required: false, added_in: v2 }
  - { name: event_type,   type: string,    required: true,  added_in: v3, default: unknown }
  - { name: country,      type: string,    required: false, added_in: v4 }
compatibility:
  backward: true
  forward:  true
deprecation_policy:
  removal_notice_minimum_days: 30
  breaking_changes_require: major-version-bump
ownership:
  team: data-engineering
  slack_channel: "#data-eng-oncall"
  primary_contact: data-engineering@example.com
sla_reference: lesson12-observability-and-slas/data_contract.yml

Step 5 — Add the new compatibility tests

Open data-engineering/sample-pipeline/schema_evolution/test_compatibility.py and append the following two tests at the bottom of the file:

def test_v4_adds_optional_country():
    """v4 schema is exactly v3 schema plus the new country column."""
    assert _schema("v4.parquet") == _schema("v3.parquet") | {"country"}


def test_v4_reads_v1_consumer_safe():
    """A v1 consumer loading v4 data must still find all v1 columns intact."""
    v4 = pq.read_table(FIXTURES / "v4.parquet").select(
        ["id", "actor", "repo", "created_at"])
    assert v4.num_rows == 10
    assert v4.column_names == ["id", "actor", "repo", "created_at"]

Step 6 — Run the suite again

cd data-engineering/sample-pipeline
pytest schema_evolution/ -v

Expected: 9 tests pass (7 original + 2 new):

test_v1_columns PASSED
test_v2_adds_optional_payload_size PASSED
test_v3_adds_required_event_type PASSED
test_v3_reads_v1_consumer_safe PASSED
test_v1_reads_v3_producer_safe PASSED
test_new_columns_have_values[v2.parquet-payload_size] PASSED
test_new_columns_have_values[v3.parquet-event_type] PASSED
test_v4_adds_optional_country PASSED
test_v4_reads_v1_consumer_safe PASSED

9 passed

Step 7 — Emit a lineage event for the v4 deploy

Use the OpenLineage helper from Lesson 12 to record the schema evolution as a pipeline job. This creates an auditable record linking events_v3 (input) to events_v4 (output):

python -c "
from observability.openlineage_emit import new_run_id, emit_start, emit_complete

rid = new_run_id()
emit_start(rid, 'evolve_events_v4',
           inputs=['events_v3'], outputs=['events_v4'])
emit_complete(rid, 'evolve_events_v4')
print('Lineage emitted. Run ID:', rid)
"

If Marquez is running (from the L12 docker-compose): open http://localhost:3000, navigate to Jobs, find evolve_events_v4, and confirm that events_v4 appears as a downstream dataset connected by the job.

If Marquez is not running (file-backend fallback): verify the two event files were written:

ls observability/events/ | grep evolve_events_v4

Expected: two files, one ending in -START.json and one ending in -COMPLETE.json.

What you should leave with

Troubleshooting