diff --git a/README.md b/README.md index 70e259b..e4b74b7 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,11 @@ python example/scripts/07_setup_grafana.py # provision monitoring dashbo See [example/README.md](example/README.md) for details. +**Delayed ground truth tutorial (POC):** a variant that trains on only a +cutoff of the data, then streams the rest in as batches with ground truth +arriving after a delay, so you can watch drift/MAE metrics evolve instead of +landing all at once. See [example/tutorial_streaming/README.md](example/tutorial_streaming/README.md). + ## Teardown ```bash diff --git a/example/tutorial_streaming/01_load_and_cutoff.py b/example/tutorial_streaming/01_load_and_cutoff.py new file mode 100644 index 0000000..3324cfa --- /dev/null +++ b/example/tutorial_streaming/01_load_and_cutoff.py @@ -0,0 +1,39 @@ +""" +Step 1: Generate the synthetic housing dataset and load only the first +TRAIN_CUTOFF_FRACTION of it into BigQuery offline_features, holding the rest +back to be streamed in later by 04_simulate_stream.py. + +Assumes a fresh mlops dataset (see README.md). +""" + +import os +from pathlib import Path +from dotenv import load_dotenv +from google.cloud import bigquery + +from _dataset import generate_rows, N_ROWS + +load_dotenv(Path.cwd() / ".env") + +PROJECT = os.environ.get("BIGQUERY_PROJECT") +if not PROJECT: + raise SystemExit( + "BIGQUERY_PROJECT is not set. Run `deployml get-urls` after `deployml deploy` to write a .env " + "with BIGQUERY_PROJECT, MLFLOW_URL, and others. Then re-run this script from the same directory." + ) +DATASET = os.getenv("BIGQUERY_DATASET", "mlops") +TABLE = f"{PROJECT}.{DATASET}.offline_features" +TRAIN_CUTOFF_FRACTION = 0.4 + +rows = generate_rows() +cutoff = int(N_ROWS * TRAIN_CUTOFF_FRACTION) +train_rows = rows[:cutoff] + +client = bigquery.Client(project=PROJECT) +errors = client.insert_rows_json(TABLE, train_rows) + +if errors: + print(f"Errors inserting rows: {errors}") +else: + print(f"✓ Loaded {len(train_rows)}/{N_ROWS} rows into {TABLE} (initial training cutoff)") + print(f" {N_ROWS - cutoff} rows held back for 04_simulate_stream.py") diff --git a/example/tutorial_streaming/04_simulate_stream.py b/example/tutorial_streaming/04_simulate_stream.py new file mode 100644 index 0000000..2ccd2ee --- /dev/null +++ b/example/tutorial_streaming/04_simulate_stream.py @@ -0,0 +1,103 @@ +""" +Step 4: Simulate new incoming data arriving in batches. Each batch is scored +via FastAPI /predict immediately, then its ground truth is written after a +short delay, so the predictions and ground_truth tables fill in at different +times like a real stream. + +ponytail: BATCH_DELAY_SECONDS is a real wall-clock sleep, not a simulated +one, so total runtime is tied to demo time. Upgrade path: write ground truth +with backdated event_timestamps instead of sleeping, so the full time series +is visible immediately regardless of how long the script takes to run. +""" + +import os +import time +from pathlib import Path +from datetime import datetime, timezone +import numpy as np +import requests +from dotenv import load_dotenv +from google.cloud import bigquery + +from _dataset import generate_rows, split_into_batches, N_ROWS + +load_dotenv(Path.cwd() / ".env") + +FASTAPI_URL = os.environ.get("FASTAPI_URL") +PROJECT = os.environ.get("BIGQUERY_PROJECT") +DATASET = os.getenv("BIGQUERY_DATASET", "mlops") +TRAIN_CUTOFF_FRACTION = 0.4 +N_BATCHES = 5 +BATCH_DELAY_SECONDS = 30 +GROUND_TRUTH_NOISE = 15000 # std dev of fake noise around predicted value + + +def score_batch(batch: list[dict]) -> int: + success = 0 + for row in batch: + payload = { + "entity_id": row["entity_id"], + "features": { + "bedrooms": row["bedrooms"], + "bathrooms": row["bathrooms"], + "area_sqft": row["area_sqft"], + "lot_size": row["lot_size"], + "year_built": row["year_built"], + "city": row["city"], + "state": row["state"], + }, + } + resp = requests.post(f"{FASTAPI_URL}/predict", json=payload, timeout=10) + if resp.status_code == 200 and resp.json().get("prediction", -1) != -1: + success += 1 + return success + + +def write_ground_truth(client: bigquery.Client, batch: list[dict]) -> None: + entity_ids = [row["entity_id"] for row in batch] + query = f""" + SELECT entity_id, predicted_value + FROM `{PROJECT}.{DATASET}.predictions` + WHERE entity_id IN UNNEST(@entity_ids) + """ + job_config = bigquery.QueryJobConfig( + query_parameters=[bigquery.ArrayQueryParameter("entity_ids", "STRING", entity_ids)] + ) + predictions = list(client.query(query, job_config=job_config).result()) + + now = datetime.now(timezone.utc) + rows = [ + { + "entity_id": row["entity_id"], + "event_timestamp": now.isoformat(), + "actual_value": float(row["predicted_value"]) + np.random.normal(0, GROUND_TRUTH_NOISE), + } + for row in predictions + ] + errors = client.insert_rows_json(f"{PROJECT}.{DATASET}.ground_truth", rows) + if errors: + print(f" Errors writing ground truth: {errors}") + else: + print(f" ✓ Wrote ground truth for {len(rows)} rows") + + +if __name__ == "__main__": + if not FASTAPI_URL or not PROJECT: + raise SystemExit( + "FASTAPI_URL or BIGQUERY_PROJECT missing. Run `deployml get-urls` to write .env." + ) + + all_rows = generate_rows() + cutoff = int(N_ROWS * TRAIN_CUTOFF_FRACTION) + stream_rows = all_rows[cutoff:] + batches = split_into_batches(stream_rows, N_BATCHES) + + client = bigquery.Client(project=PROJECT) + for i, batch in enumerate(batches, start=1): + print(f"Batch {i}/{len(batches)}: scoring {len(batch)} rows") + success = score_batch(batch) + print(f" ✓ {success}/{len(batch)} predictions successful") + + print(f" Waiting {BATCH_DELAY_SECONDS}s before ground truth arrives...") + time.sleep(BATCH_DELAY_SECONDS) + write_ground_truth(client, batch) diff --git a/example/tutorial_streaming/README.md b/example/tutorial_streaming/README.md new file mode 100644 index 0000000..38ddfbe --- /dev/null +++ b/example/tutorial_streaming/README.md @@ -0,0 +1,86 @@ +# Delayed Ground Truth Tutorial (POC) + +A small variant of the [end-to-end example](../README.md) that simulates new +incoming data arriving over time, with ground truth showing up after a delay +instead of all at once. Built for issue #65. + +**Assumption:** this expects a fresh `mlops` BigQuery dataset (no rows yet in +`offline_features`, `predictions`, `ground_truth`). If you already ran the +main `example/scripts/` walkthrough against the same deployment, either +`deployml destroy` + `deployml deploy` again, or point this at a separate +deployment. + +## Prerequisites + +Same as the [main example](../README.md#prerequisites): a deployed stack, +`.env` from `deployml get-urls`, and the same Python dependencies. + +## Steps + +Run from the project root (where `.env` lives). + +**1. Load training data with a cutoff** + +```bash +python example/tutorial_streaming/01_load_and_cutoff.py +``` + +Generates the same 500-row synthetic housing dataset as the main example, but +only loads the first 40% (200 rows) into `offline_features`. The remaining +300 rows are held back to be streamed in later. + +**2. Train a model** + +```bash +python example/scripts/02_train_model.py +``` + +Unchanged from the main example. Trains on whatever is in `offline_features` +— now just the 200-row cutoff instead of the full 500. + +**3. Register the model** + +```bash +python example/scripts/03_register_model.py +``` + +Unchanged from the main example. + +**4. Simulate the incoming data stream** + +```bash +python example/tutorial_streaming/04_simulate_stream.py +``` + +Regenerates the same 500 rows (same seed as step 1) and takes the 300 held +back. Splits them into 5 batches of ~60 rows. For each batch: scores it via +FastAPI `/predict` (logged to `predictions` automatically), waits 30 seconds, +then writes ground truth for that batch to `ground_truth`. Takes about 2.5 +minutes total. + +**5. Compute drift metrics** + +```bash +python example/scripts/06_compute_drift_metrics.py +``` + +Unchanged from the main example. + +**6. Set up the Grafana dashboard** + +```bash +python example/scripts/07_setup_grafana.py +``` + +Unchanged from the main example. Open `GRAFANA_URL` to watch predictions and +MAE update batch by batch if you run step 4 again, or re-run step 5 between +batches. + +## Known limitations / future work + +- The 30-second delay is a real `time.sleep`, not a simulated one, so it's + tied to wall-clock demo time. A follow-up will replace it with backdated + `event_timestamp`s so a full time series is visible immediately. +- This is a POC run manually as scripts. A follow-up will promote it into a + `deployml tutorial` CLI command, including support for uploading your own + dataset instead of the built-in housing one. diff --git a/example/tutorial_streaming/_dataset.py b/example/tutorial_streaming/_dataset.py new file mode 100644 index 0000000..13a3c35 --- /dev/null +++ b/example/tutorial_streaming/_dataset.py @@ -0,0 +1,50 @@ +"""Shared synthetic housing data generator for the streaming tutorial. + +Same feature distributions and formula as example/scripts/01_load_training_data.py, +factored out so 01_load_and_cutoff.py and 04_simulate_stream.py can both +regenerate the identical 500 rows (same seed) without persisting anything +to disk between script runs. +""" + +import uuid +from datetime import datetime, timezone +import numpy as np + +N_ROWS = 500 +RANDOM_SEED = 42 + + +def generate_rows(n_rows: int = N_ROWS, seed: int = RANDOM_SEED) -> list[dict]: + rng = np.random.default_rng(seed) + + cities = list(range(5)) # 0-4 representing 5 cities + states = list(range(3)) # 0-2 representing 3 states + + bedrooms = rng.integers(1, 6, n_rows).astype(float) + bathrooms = rng.integers(1, 4, n_rows).astype(float) + area_sqft = rng.integers(800, 4000, n_rows).astype(float) + lot_size = rng.integers(2000, 10000, n_rows).astype(float) + year_built = rng.integers(1960, 2023, n_rows).astype(float) + city = rng.choice(cities, n_rows).astype(float) + state = rng.choice(states, n_rows).astype(float) + + now = datetime.now(timezone.utc) + return [ + { + "entity_id": str(uuid.uuid4()), + "event_timestamp": now.isoformat(), + "bedrooms": bedrooms[i], + "bathrooms": bathrooms[i], + "area_sqft": area_sqft[i], + "lot_size": lot_size[i], + "year_built": year_built[i], + "city": city[i], + "state": state[i], + } + for i in range(n_rows) + ] + + +def split_into_batches(rows: list, n_batches: int) -> list[list]: + """Split rows into n_batches near-equal chunks, covering every row exactly once.""" + return [list(batch) for batch in np.array_split(rows, n_batches)] diff --git a/tests/test_tutorial_streaming.py b/tests/test_tutorial_streaming.py new file mode 100644 index 0000000..27d8abc --- /dev/null +++ b/tests/test_tutorial_streaming.py @@ -0,0 +1,41 @@ +"""Unit test for the batch-splitting logic in the streaming tutorial. No GCP/FastAPI calls.""" +import importlib.util +from pathlib import Path + +DATASET_MODULE_PATH = Path(__file__).parent.parent / "example" / "tutorial_streaming" / "_dataset.py" + + +def _load_dataset_module(): + spec = importlib.util.spec_from_file_location("tutorial_streaming_dataset", DATASET_MODULE_PATH) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +split_into_batches = _load_dataset_module().split_into_batches + + +def test_split_into_batches_covers_every_row_once(): + rows = [{"entity_id": i} for i in range(300)] + batches = split_into_batches(rows, 5) + + assert len(batches) == 5 + flattened = [row for batch in batches for row in batch] + assert sorted(row["entity_id"] for row in flattened) == list(range(300)) + + +def test_split_into_batches_near_equal_sizes(): + rows = [{"entity_id": i} for i in range(7)] + batches = split_into_batches(rows, 3) + + sizes = sorted(len(b) for b in batches) + assert sizes == [2, 2, 3] + + +def test_split_into_batches_more_batches_than_rows(): + rows = [{"entity_id": i} for i in range(2)] + batches = split_into_batches(rows, 5) + + assert len(batches) == 5 + flattened = [row for batch in batches for row in batch] + assert sorted(row["entity_id"] for row in flattened) == [0, 1]