diff --git a/.gitignore b/.gitignore index 7ca2f44..bcf506f 100644 --- a/.gitignore +++ b/.gitignore @@ -16,3 +16,6 @@ package-lock.json *.csv *.swp pipeline.json +.df-credentials.json +content/agenticdata/seed_data/ +seed_data/ diff --git a/README.md b/README.md index 3389cb2..0084b13 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,7 @@ Bootkon consits of the following streams: | ------------- | ----------- | -------------- | | [Data & AI](content/data/README.md) | Your role: As a senior data analytics/AI engineer at an imaginary company called FraudFix Technologies, you will tackle the challenges of making financial transactions safer using machine learning. Your work will involve analyzing vast amounts of transaction data to detect and prevent fraud, as well as assessing customer sentiment regarding the quality of transaction services. You will leverage a unique dataset, which includes auto-generated data by Google Gemini and public European credit card transactions that have been PCA transformed and anonymized. This dataset will be used to train your models, reflecting real-world applications of GCP Data & AI in enhancing financial safety. | [Fabian Hirschmann](https://www.linkedin.com/in/fhirschmann/) (maintainer; main author), [Cary Edwards](https://www.linkedin.com/in/cary-edwards-a3a557a6/) (contributor), [Wissem Khlifi](https://www.linkedin.com/in/orawiss/) (original author), [Daniel Quinlan](https://www.linkedin.com/in/%F0%9F%8C%8Ddaniel-quinlan-51126016/), [Daniel Holgate](https://www.linkedin.com/in/danielholgate/) | | [DevEx & AI](content/devex/README.md) | Your role: As a senior AI powered developer/engineer, the goal is on utilizing Antigravity CLI for accelerating the execution of various tasks, including understanding existing codebases, generating documentation and unit tests, refactoring both UI and backend components of a Python web application and deploying the application to Cloud Native Runtimes like Cloud Run.| [Kaushik Bhattacharya](https://www.linkedin.com/in/kaushikbhattacharya/) (maintainer; main author) | +| [Agentic Data](content/agenticdata/README.md) | Your role: As a data platform engineer at Cymbal, a fictional commerce company, you build an end-to-end agentic data platform in one afternoon: the operational PostgreSQL database is replicated live into BigQuery with Datastream (CDC over Private Service Connect), refined into a governed bronze/silver/gold medallion with Dataform — where the Antigravity CLI (agy) authors the SQL under your direction — governed with Knowledge Catalog, and topped by a BigQuery conversational data agent that an ADK concierge agent consumes over the A2A protocol, alongside live order lookups against Cloud SQL. The dataset is fully synthetic and generated inside your own project. | [Fabian Hirschmann](https://www.linkedin.com/in/fhirschmann/) (maintainer; main author), [Florian Baumert](https://www.linkedin.com/in/florian-baumert/), [Cary Edwards](https://www.linkedin.com/in/cary-edwards-a3a557a6/) | ## Note diff --git a/content/agenticdata/README.md b/content/agenticdata/README.md new file mode 100644 index 0000000..6b4c442 --- /dev/null +++ b/content/agenticdata/README.md @@ -0,0 +1,185 @@ +# Agentic Data Bootkon + +Welcome to Agentic Data Bootkon! + +Agentic Data Bootkon is an immersive half-day hackathon for data engineers, architects, and AI enthusiasts. In one afternoon (13:00–17:00) you build a complete **agentic data platform** on Google Cloud — from a live operational database all the way to AI agents that talk to each other about your data over an open protocol. Throughout the event, the **Antigravity CLI (`agy`)** works alongside you as an AI co-engineer: you run the infrastructure, agy writes the code, and the Cloud Console is where you verify what happened. + +This event is comprised of the following code labs: + +| Time | Duration | Topic | Details | +| --- | --- | --- | --- | +| 13:00 | 10min | Kickoff | Credentials, Cloud Shell, tutorial setup | +| 13:10 | 20min | Environment Setup & agy | IAM, network, async database build, seed data, meet your AI co-engineer | +| 13:30 | 40min | Live Replication | Cloud SQL for PostgreSQL → Datastream CDC → BigQuery bronze, with a live activity simulator | +| 14:10 | 40min | Dataform with agy | agy authors the bronze→silver→gold medallion pipeline; you compile, run, and verify it | +| 14:50 | 30min | Break | ☕ | +| 15:20 | 30min | Data Governance | Knowledge Catalog: tier aspects, auto data quality, PII policy tags, glossary, lineage | +| 15:50 | 20min | BigQuery Data Agent | A conversational analytics agent grounded on your governed gold layer | +| 16:10 | 45min | A2A Finale | Two ADK agents — analytics via A2A, live order lookups via Cloud SQL — in one conversation | +| 16:55 | 5min | Wrap-up | 🎉 | + +You can navigate this handbook using the `<` and `>` buttons on the right and left hand side, respectively. To get started, please press the `>` button on the right hand side now. + +## Use Case + +Your role: you are a data platform engineer at **Cymbal**, a commerce company whose order platform runs on PostgreSQL. The analytics team is drowning in one-off CSV exports, nobody trusts the numbers, and management wants "AI on our data" — safely. Over one afternoon, you will give Cymbal: + +1. **A live replica instead of exports.** Datastream captures every insert, update, and delete from Cloud SQL for PostgreSQL and merges it into BigQuery within moments — over Private Service Connect, with no public IPs anywhere. +2. **Numbers people can trust.** A medallion architecture (bronze → silver → gold) built with Dataform, where the SQL is authored by the Antigravity CLI under your direction, complete with data tests. The raw data contains real-world-style flaws (duplicates, typos, orphaned records) — cleaning them is the point. +3. **Governance before agents.** Knowledge Catalog labels the tiers, scans quality continuously, locks down PII with enforced column-level security, and gives business terms a home — the grounding layer that makes AI answers trustworthy. +4. **Agents on top.** A BigQuery conversational data agent over the gold layer, then an ADK multi-agent system: a concierge agent that answers live operational questions straight from Postgres and delegates analytical questions to the data agent over the **A2A protocol** — *agent-to-agent*, the open standard (governed by the Linux Foundation) that lets AI agents from any framework or vendor discover and call each other, the way HTTP lets services talk to each other. + +## Architecture + + + + + + + + + + + + + + + + + + + + + + + + + +
+ Cloud SQL
+ Cloud SQL
+ PostgreSQL, PSC-only
Cymbal's live order DB + 		— CDC ▶ + Datastream
+ Datastream
+ backfill + streaming
merge mode + 		—▶ + BigQuery
+ BigQuery
+ bronze → silver → gold
Dataform SQLX, authored by agy + 		—▶ + BigQuery data agent
+ Data agent
+ conversational analytics
grounded on gold +
+ ▲
+ Knowledge Catalog
+ Knowledge Catalog
+ tier aspects · data quality
PII policy tags · glossary · lineage + 		▲
A2A
+ ▲
+ live order
lookups + 		◀———————————————— + ADK concierge agent
+ cymbal_concierge
+ ADK agent, adk web +
+ Cloud Shell
+ Cloud Shell + Antigravity CLI (agy)
+ you operate the shell · agy authors the code · the console verifies +
+ +## About the data set + +There is no download: the **Cymbal orders** dataset is fully synthetic and generated *inside your own project* by a seeded generator — every participant gets identical data. It models a generic commerce platform (customers, products, orders, order items, payments, reviews; ~2.3M rows, well under 2 GB) and includes deliberately planted data-quality flaws — duplicate customers, invalid emails, a `shiped` status typo, mixed-case currencies, orphaned line items, negative payments, future timestamps — plus PII columns (names, emails, phones, addresses) for the governance lab. A simulator keeps inserting and updating rows during the event, so the change-data-capture pipeline always has something to show. + +## Logging into Google Cloud + +> [!CAUTION] +> Please follow the below steps exactly as written. Deviating from them has unintended consequences. + +Let us set up your Google Cloud Console. Please: + +1. Open a new browser window in **Incognito** mode. +2. Open this handbook in your newly opened incognito window and keep reading; close this window in your main browser window. +3. Open Google Cloud Console and log in with the provided credentials. +4. Accept the Terms of Services. + + ![](../common/img/termsofservice.png) + +5. Choose your **project id**. Click on select a project and select the project ID (example below) + ![](../common/img/selectproject.png) + + + ![](../common/img/selectproject2.png) + + + ![](../common/img/selectproject3.png) + +6. Go to [language settings](https://console.cloud.google.com/user-preferences/languages) and change your language to `English (US)`. This will help our tutorial engine recognize items on your screen and make our table captain be able to help you. + + ![](../common/img/select_language.png) + +## Executing code labs + +During this event, we will guide you through a series of labs using Google Cloud Shell. + +Cloud Shell is a fully interactive, browser-based environment for learning, experimenting, and managing Google Cloud projects. It comes preloaded with the Google Cloud CLI, essential utilities, and a built-in code editor with Cloud Code integration, enabling you to develop, debug, and deploy cloud apps entirely in the cloud. + +Below you can find a screenshot of Cloud Shell. + +![](../common/img/cloud_shell_window.png) + +It is based on Visual Studio Code and hence looks like a normal IDE. However, on the right hand side you see the tutorial you will be working through. When you encounter code chunks in the tutorial, there are two icons on the right hand side. One to copy the code chunk to your clipboard and the other one to insert it directly into the terminal of Cloud Shell. + +## Working with labs (important) + +> [!CAUTION] +> Please note the points in this section before you get started with the labs in the next section. + +While going through the code labs, you will encounter two different terminals on your screen. Please only use the terminal from the IDE (white background) and do not use the non-IDE terminal (black background). In fact, just close the terminal with black background using the `X` button. + +![](../common/img/code_terminals.png) + +You will also find two buttons on your screen that might seem tempting. Please do not click the *Open Terminal* or *Open in new window* buttons as they will destroy the integrated experience of Cloud Shell. + +![](../common/img/code_newwindow.png) + +Please double check that the URL in your browser reads `console.cloud.google.com` and not `shell.cloud.google.com`. + +![](../common/img/wrong_url.png) + +Should you accidentally close the tutorial or the IDE, just type the following command into the terminal: + +```bash +bk-start +``` + +> [!NOTE] +> This stream uses several terminal tabs at once — an IAP tunnel and a data simulator keep running in the background. Don't close those tabs; new tabs are ready to use right away (your configuration loads automatically from `~/.bashrc`). + +## Start the lab + +In your Google Cloud Console window, activate Cloud Shell. + +![](../common/img/activate_cloud_shell.png) + +Click into the terminal that has opened at the bottom of your screen. + +![](../common/img/cloud_shell_terminal.png) + +And copy & paste the following command and press return: + +```bash +BK_STREAM=agenticdata BK_REPO=GoogleCloudPlatform/bootkon; . <(wget -qO- https://raw.githubusercontent.com/${BK_REPO}/main/.scripts/bk) +``` + +Now, please go back to Cloud Shell and continue with the tutorial that has been opened on the right hand side of your screen! + + +## Authors + +The authors of Agentic Data Bootkon are: +- [Fabian Hirschmann](https://www.linkedin.com/in/fhirschmann/) (maintainer; main author) +- [Florian Baumert](https://www.linkedin.com/in/florian-baumert/) +- [Cary Edwards](https://www.linkedin.com/in/cary-edwards-a3a557a6/) diff --git a/content/agenticdata/TUTORIAL.md b/content/agenticdata/TUTORIAL.md new file mode 100644 index 0000000..b4ffbfd --- /dev/null +++ b/content/agenticdata/TUTORIAL.md @@ -0,0 +1,87 @@ + + + + + + + + +# Agentic Data Bootkon + + +Author: Fabian Hirschmann + + + +## Introduction + +{% if MY_NAME %}Hi **{{ MY_NAME }}! 👋** {% endif %}Welcome to Agentic Data Bootkon. We're delighted to have you! This sidebar contains parts of the labs you will work through. Before we get started, let's set up a few things: + +### Working with labs + +You can insert commands into the terminal using the following button on top of each code line in the tutorial: +. The button will automatically open the terminal. +Please make sure you are using the terminal of the IDE. + +Let's try: + +```bash +echo "Let's go, Cymbal!" +``` + +Execute by pressing the return key in the terminal that has been opened in the lower part of your screen. + +### Setting environment variables +You can open files directly from this tutorial. +Open `vars.sh` by clicking here +and set `GCP_USERNAME`, `PROJECT_ID` according to the information you received. Also let us know your (first?) name in `MY_NAME`. Don't forget to save it. + +❗ Please do not include any whitespaces when setting these variables. + +Please initialize bootkon. The next command will set environment variables in your current terminal. + +```bash +. bk +``` + +Reload the tutorial window on the right-hand side of your screen. + +```bash +bk-start +``` + +In case you accidentally close the tutorial or the editor, you can execute `bk-start` to start it again. Please make sure that you execute `. bk` in every terminal +you open so that the environment variables are set. + +Now, your + +* `PROJECT_ID` is `{% if PROJECT_ID == "" %}None{% else %}{{ PROJECT_ID }}{% endif %}` + +* `GCP_USERNAME` is `{% if GCP_USERNAME == "" %}None{% else %}{{ GCP_USERNAME }}{% endif %}`. + + +If neither is `None`, press the `START` button below to get started! + +{% include 'agenticdata/labs/1_environment_setup.md' %} + +{% include 'agenticdata/labs/2_replication.md' %} + +{% include 'agenticdata/labs/3_dataform.md' %} + +{% include 'agenticdata/labs/4_governance.md' %} + +{% include 'agenticdata/labs/5_data_agent.md' %} + +{% include 'agenticdata/labs/6_a2a.md' %} + + +## The end + +You've made it! Congratulations and thank you for participating in this event. + +### Credits + +The authors of Agentic Data Bootkon are: +- [Fabian Hirschmann](https://www.linkedin.com/in/fhirschmann/) (main author) +- [Florian Baumert](https://www.linkedin.com/in/florian-baumert/) +- [Cary Edwards](https://www.linkedin.com/in/cary-edwards-a3a557a6/) diff --git a/content/agenticdata/bk-bootstrap b/content/agenticdata/bk-bootstrap new file mode 100755 index 0000000..d45c7dd --- /dev/null +++ b/content/agenticdata/bk-bootstrap @@ -0,0 +1,168 @@ +#!/bin/bash +# Bootstraps the Agentic Data Bootkon stream: Python deps, APIs, IAM, +# service accounts, and the secrets used across the labs. +# +# It deliberately does NOT create the Cloud SQL instance or any networking: +# those commands live inline in the labs, so the connectivity variant can be +# swapped by applying option1-vpc-peering-proxy.diff to the lab markdown. + +[[ "$BK_INITIALIZED" != "1" ]] && echo "Bootkon has not been initialized. Please follow the tutorial and run '. bk'" && exit 1 + +# The Datastream service agent must exist and hold its service-agent role +# BEFORE Lab 1 creates the private connection: on a fresh project the agent +# is created lazily, and `private-connections create` then fails with +# PERMISSION_DENIED on the network attachment (verified live). The IAM grant +# also needs a few minutes to propagate -- doing this FIRST, before the +# pip install and the role loop below, buys that time for free. +gcloud services enable datastream.googleapis.com >/dev/null 2>&1 +gcloud beta services identity create --service=datastream.googleapis.com --project=$PROJECT_ID >/dev/null 2>&1 +gcloud projects add-iam-policy-binding "$PROJECT_ID" \ + --member="serviceAccount:service-${PROJECT_NUMBER}@gcp-sa-datastream.iam.gserviceaccount.com" \ + --role="roles/datastream.serviceAgent" --condition=None >>/dev/null + +pip install -r content/agenticdata/requirements.txt + +echo "Enabling APIs (idempotent; the tutorial enables these too) ..." +gcloud services enable \ + sqladmin.googleapis.com \ + datastream.googleapis.com \ + compute.googleapis.com \ + servicenetworking.googleapis.com \ + iap.googleapis.com \ + bigquery.googleapis.com \ + bigqueryconnection.googleapis.com \ + dataplex.googleapis.com \ + datacatalog.googleapis.com \ + datalineage.googleapis.com \ + dlp.googleapis.com \ + geminidataanalytics.googleapis.com \ + cloudaicompanion.googleapis.com \ + aiplatform.googleapis.com \ + storage-component.googleapis.com \ + serviceusage.googleapis.com \ + cloudresourcemanager.googleapis.com \ + iam.googleapis.com \ + artifactregistry.googleapis.com + +# Knowledge Catalog data-quality service account: profile/quality scans in +# Lab 4 run as this SA (custom execution identity). +DQ_SCAN_SERVICE_ACCOUNT="dataquality-service-account" +DQ_SCAN_SA_EMAIL="${DQ_SCAN_SERVICE_ACCOUNT}@${PROJECT_ID}.iam.gserviceaccount.com" +gcloud iam service-accounts create $DQ_SCAN_SERVICE_ACCOUNT --display-name="Knowledge Catalog DQ Service Account" +# A freshly created SA can take a few seconds to become visible to IAM; +# granting against it too early fails with "does not exist". +for i in $(seq 1 15); do + gcloud iam service-accounts describe "$DQ_SCAN_SA_EMAIL" >/dev/null 2>&1 && break + sleep 2 +done + +# Roles for the participant. Sandbox users are typically project owners +# already; the explicit list keeps the stream portable to other setups. +declare -a user_roles=( + "roles/cloudsql.admin" # create/manage the Cymbal instance + "roles/datastream.admin" # profiles, private connection, stream + "roles/compute.admin" # VPC, jump VM, firewall, PSC endpoint + "roles/iap.tunnelResourceAccessor" # IAP TCP tunnel to the jump VM + "roles/bigquery.admin" # datasets, jobs, connections + "roles/dataplex.admin" # Knowledge Catalog: aspects, scans, glossary + "roles/datalineage.admin" # lineage graph + "roles/aiplatform.user" # Gemini calls from ADK agents + "roles/geminidataanalytics.dataAgentCreator" # create + chat with data agents + "roles/cloudaicompanion.user" # conversational analytics chat + "roles/storage.admin" # seed bucket + "roles/iam.serviceAccountAdmin" # create service accounts + "roles/iam.serviceAccountUser" # act as service accounts +) + +for role in "${user_roles[@]}"; do + echo "Assigning role $role to $GCP_USERNAME in project $PROJECT_ID..." + gcloud projects add-iam-policy-binding "$PROJECT_ID" \ + --member="user:$GCP_USERNAME" --role="$role" >>/dev/null +done + +# Roles for the data-quality service account. +declare -a dataquality_roles=( + "roles/bigquery.jobUser" + "roles/bigquery.dataViewer" + "roles/iam.serviceAccountTokenCreator" + "roles/iam.serviceAccountUser" + "roles/storage.objectViewer" +) + +for role in "${dataquality_roles[@]}"; do + echo "Assigning role $role to $DQ_SCAN_SA_EMAIL in project $PROJECT_ID..." + gcloud projects add-iam-policy-binding "$PROJECT_ID" \ + --member="serviceAccount:$DQ_SCAN_SA_EMAIL" --role="$role" >>/dev/null +done + +# Per-participant configuration, appended to ~/.bashrc so every new terminal +# picks it up automatically (same pattern .scripts/bk uses for BK_* vars). +if ! grep -q '# agenticdata-bootkon' ~/.bashrc; then + echo "Generating database passwords and adding stream exports to ~/.bashrc ..." + cat >> ~/.bashrc < "$AGY_DIR/settings.json" < +{{ author('Fabian Hirschmann', 'https://linkedin.com/in/fhirschmann') }} + + + +Welcome to Cymbal{% if MY_NAME %}, {{ MY_NAME }}{% endif %}! By 17:00 you will have built a complete agentic data platform: a live operational database, change-data-capture into BigQuery, a governed bronze→silver→gold architecture, and two AI agents talking to each other over **A2A** — the open *agent-to-agent* protocol that lets AI agents call each other the way HTTP lets services do it. Don't worry if that's new to you: the finale lab explains and builds it step by step. + +In this lab you will enable services, kick off the two slow infrastructure builds (they run in the background while you work), stage the seed data, and meet **Antigravity CLI (`agy`)** — your co-engineer for the afternoon. + +One rule for today: **you** run the infrastructure commands, **agy** writes code and configs, and the **console** is where you verify what happened. + +### Enable services + + + + +### Assign permissions + +Execute the following script. It installs Python dependencies, grants IAM roles, creates a service account for data-quality scans, and generates the database passwords for your project. It runs for about two minutes — you can inspect bk-bootstrap while it works: + +```bash +content/agenticdata/bk-bootstrap +``` + +The script appended your generated database passwords and agent settings to `~/.bashrc`, so **every new terminal picks them up automatically**. Load them into this already-open one: + +```bash +source ~/.bashrc +``` + +Then reload the tutorial once — this renders your freshly generated passwords directly into the commands of the next labs: + +```bash +bk-start +``` + +### Create the network path + +Our database will have **no public IP**. Instead, Datastream will reach it through Private Service Connect (PSC). First, create a VPC and a subnet: + +```bash +gcloud compute networks create cymbal-vpc --subnet-mode=custom +``` + +```bash +gcloud compute networks subnets create cymbal-subnet \ + --network=cymbal-vpc --region={{ REGION }} --range=10.10.0.0/24 +``` + +Datastream connects into your VPC through a **network attachment** — the entry door for its PSC interface: + +```bash +gcloud compute network-attachments create cymbal-attachment --region={{ REGION }} \ + --connection-preference=ACCEPT_AUTOMATIC --subnets=cymbal-subnet +``` + +Note: `ACCEPT_AUTOMATIC` keeps this workshop simple. In production you would use `ACCEPT_MANUAL` with a `--producer-accept-list`, discovering the Datastream tenant project via `gcloud datastream private-connections create --validate-only`. + +So what did you just build? **Private Service Connect** is Google Cloud's way of exposing a service across VPC boundaries without peering entire networks or using public IPs: the producer (here: your Cloud SQL instance) publishes a *service attachment*, and consumers plug an *endpoint* into it — traffic stays on Google's backbone, and each side only sees the single socket it was given. The **network attachment** you created is the reverse construct: it lets a Google-managed producer (Datastream) place a network interface *into* your VPC. You will connect both halves in Lab 2. + +Learn more: +- [Private Service Connect](https://docs.cloud.google.com/vpc/docs/private-service-connect) +- [Network attachments](https://docs.cloud.google.com/vpc/docs/about-network-attachments) +- [Cloud SQL and Private Service Connect](https://docs.cloud.google.com/sql/docs/postgres/about-private-service-connect) + +### Launch your operational database + +This is Cymbal's production order database: PostgreSQL on Cloud SQL, with logical decoding enabled at creation time (Datastream needs it for CDC) and Private Service Connect instead of a public IP. The `--async` flag returns immediately — the instance builds in the background for the next 10–15 minutes while you continue: + +{% if ON_ARGOLIS %} +❗ You are on Argolis. Instance and VM creation in this stream may be blocked by organization policies (for example `constraints/compute.requireOsLogin` or `constraints/compute.requireShieldedVm`). If a create command fails with a policy error, disable the constraint under IAM & Admin → Organization Policies and retry. +{% endif %} + +```bash +gcloud sql instances create cymbal-oltp \ + --database-version=POSTGRES_15 --edition=enterprise \ + --tier=db-custom-1-3840 --storage-size=10GB \ + --region={{ REGION }} \ + --root-password=$BK_DB_PASSWORD \ + --database-flags=cloudsql.logical_decoding=on \ + --enable-private-service-connect \ + --allowed-psc-projects={{ PROJECT_ID }} \ + --no-assign-ip \ + --async +``` + +A word on the flags: `cloudsql.logical_decoding=on` switches Postgres' write-ahead log to *logical* decoding — the prerequisite for change data capture — set at creation time so the instance boots with it and never needs a flag-change restart. `--no-assign-ip` means there is no public address at all; the remaining flags wire the instance to the private path you prepared above. And `db-custom-1-3840` is a deliberately small machine: CDC reads the log, it does not stress the database. + +Learn more: +- [Set up logical replication on Cloud SQL](https://docs.cloud.google.com/sql/docs/postgres/replication/configure-logical-replication) + +### Kick off Datastream private connectivity + +Datastream's private connection takes around 5–10 minutes to build, so kick it off now too. The command returns right away and the build continues in the background — Lab 2 verifies it reached the `CREATED` state: + +```bash +gcloud datastream private-connections create cymbal-psc --location={{ REGION }} \ + --display-name=cymbal-psc \ + --network-attachment=projects/{{ PROJECT_ID }}/regions/{{ REGION }}/networkAttachments/cymbal-attachment +``` + +### Stage the seed data + +Cymbal's order history is generated **inside your project** — deterministic synthetic data, so every participant works with identical rows (including some deliberately broken ones you will meet again in Lab 3). Create a bucket and generate the data: + +```bash +gcloud storage buckets create gs://{{ PROJECT_ID }}-bucket --location={{ REGION }} +``` + +```bash +python3 content/agenticdata/src/datagen/generate.py --out seed_data +``` + +While it runs (about a minute), have a look at generate.py — note the *planted flaws* section at the top. Then upload the CSVs: + +```bash +gcloud storage cp seed_data/*.csv gs://{{ PROJECT_ID }}-bucket/seed/ +``` + +### Meet your co-engineer + +`agy` (Antigravity CLI) is already installed in Cloud Shell, and `bk-bootstrap` just pre-configured it for you: it signs in with your ambient Cloud Shell credentials and reads your project (`{{ PROJECT_ID }}`) and `global` location from the environment — so there is no login wizard to click through. Start it from the repository root (you are already there): + +```bash +agy +``` + +On a very first run it may ask you to accept the Terms of Service — do so if prompted; otherwise you land straight at the agy prompt. + +`bk-bootstrap` also tuned agy's permissions for you: it **auto-approves file edits** in the workshop folders (so authoring flows without a prompt on every file), but still **asks before running any command** — so you stay in control of anything that touches your cloud resources. Reviewing what agy did (with `/diff`) stays your job. + +### Let agy explain what just happened + +Your first prompt. Paste this into agy: + +``` +What does content/agenticdata/bk-bootstrap do? Summarize the IAM roles it grants and explain why a separate data-quality service account is created. +``` + +Read the answer — this is the pattern for the whole afternoon: you stay in command, agy does the reading and writing, and you verify. + +### Verify in the console + +Let's check on your database build. Open [Cloud SQL instances](https://console.cloud.google.com/sql/instances) and find cymbal-oltp. You should see it being created (a spinner) or already running with **Private service connect** as its connectivity — and no public IP anywhere. If it is still creating, perfect: that's exactly why we started it first. + +### Success + +🎉 Congratulations{% if MY_NAME %}, {{ MY_NAME }}{% endif %}! Your project has IAM sorted, a VPC with a door for Datastream, a database building itself in the background, half a million synthetic orders staged in Cloud Storage, and an AI co-engineer standing by in your terminal. On to the data! 🚀 diff --git a/content/agenticdata/labs/2_replication.md b/content/agenticdata/labs/2_replication.md new file mode 100644 index 0000000..a247d48 --- /dev/null +++ b/content/agenticdata/labs/2_replication.md @@ -0,0 +1,273 @@ +## Lab 2: Live Replication with Datastream + + +{{ author('Fabian Hirschmann', 'https://linkedin.com/in/fhirschmann') }} + + + +In this lab you bring Cymbal's database to life and replicate it — continuously — into BigQuery using **Datastream** change data capture (CDC). By the end, every INSERT, UPDATE and DELETE in Postgres lands in your `cymbal_bronze` dataset within moments. + +(The commands below contain your generated database passwords, rendered into the tutorial. If you see empty quotes instead, run the `bk-start` reload step from the end of Lab 1.) + +### About Datastream + +Datastream is Google Cloud's serverless **change data capture (CDC)** service: instead of periodically re-exporting tables, it taps the source database's transaction log and replicates every INSERT, UPDATE and DELETE as it happens. For PostgreSQL that log is the **write-ahead log (WAL)**: Postgres emits every committed change in *logical* form, and Datastream consumes that feed through a replication slot. A stream has two phases that run in parallel — a one-time **backfill** (snapshot of the existing rows) and continuous **CDC streaming** — and the destination side writes into BigQuery for you: no pipeline code, no cluster to manage. + +Learn more: +- [Datastream overview](https://docs.cloud.google.com/datastream/docs/overview) +- [PostgreSQL as a source](https://docs.cloud.google.com/datastream/docs/sources-postgresql) + +### Wait for the database + +If the Lab 1 build already finished, this returns instantly — otherwise it waits for the create operation to complete: + +```bash +gcloud sql operations wait --timeout=unlimited \ + $(gcloud sql operations list --instance=cymbal-oltp --format='value(name)' --limit=1) +``` + +Also confirm the Datastream private connection from Lab 1 reached the `CREATED` state (it takes around 5–10 minutes from when you kicked it off — if it still shows `CREATING`, wait a moment and re-run): + +```bash +gcloud datastream private-connections describe cymbal-psc --location={{ REGION }} --format='value(state)' +``` + +❗ If this prints `NOT_FOUND` or `FAILED`, the creation lost a race against IAM propagation for the freshly created Datastream service agent. No harm done — just re-run the `gcloud datastream private-connections create` command from Lab 1 (delete the failed one first if `FAILED`), and it will go through. + +### Create the Private Service Connect endpoint + +Your instance exposes a **service attachment** — a private socket other networks can plug into. Create an endpoint for it in your VPC at the reserved IP `10.10.0.5`: + +```bash +SA_URI=$(gcloud sql instances describe cymbal-oltp --format="value(pscServiceAttachmentLink)") +gcloud compute addresses create cymbal-endpoint-ip --region={{ REGION }} \ + --subnet=cymbal-subnet --addresses=10.10.0.5 +gcloud compute forwarding-rules create cymbal-endpoint --region={{ REGION }} \ + --address=cymbal-endpoint-ip --network=cymbal-vpc \ + --target-service-attachment=$SA_URI --allow-psc-global-access +``` + +From now on, `10.10.0.5` **is** your database — for Datastream and for the jump VM below. + +### Create the jump VM + +Cloud Shell lives outside your VPC and a PSC-only instance has no public IP, so you need a tiny helper: an e2-micro VM (no external IP either!) that forwards port 5432 to the database endpoint. Its startup script is four lines of iptables. You will reach the VM through an **IAP tunnel** — identity-based, no IPs exposed anywhere: + +```bash +gcloud compute instances create cymbal-jump --zone={{ REGION }}-a \ + --machine-type=e2-micro --subnet=cymbal-subnet --no-address --can-ip-forward \ + --metadata-from-file=startup-script=content/agenticdata/src/jumpvm-startup.sh +``` + +```bash +gcloud compute firewall-rules create allow-iap-ingress --network=cymbal-vpc \ + --direction=INGRESS --action=allow --rules=tcp:22,tcp:5432 \ + --source-ranges=35.235.240.0/20 +``` + +### Open the tunnel + +Open a **second terminal tab** (`+`) and start the tunnel. **Leave this terminal open for the rest of the event** — the simulator and (much later) your concierge agent use it. The helper wraps `gcloud compute start-iap-tunnel` and reconnects automatically if the tunnel ever drops: + +```bash +~/bootkon/content/agenticdata/bk-tunnel +``` + +When you see *Listening on port [5432]*, `localhost:5432` in Cloud Shell is your production database. + +**Identity-Aware Proxy (IAP)** TCP forwarding is what makes this safe: the tunnel is authorized by your Google identity and an IAM role (`iap.tunnelResourceAccessor`), not by network position — no VPN, no bastion with a public IP, and every connection is auditable. ([IAP TCP forwarding](https://docs.cloud.google.com/iap/docs/using-tcp-forwarding)) + +### Create the schema and load the data + +Back in your **first terminal**. Create the database (control-plane, no tunnel needed): + +```bash +gcloud sql databases create cymbal --instance=cymbal-oltp +``` + +Apply the schema through the tunnel: + +```bash +PGPASSWORD="{{ BK_DB_PASSWORD }}" psql -h localhost -p 5432 -U postgres -d cymbal \ + -f content/agenticdata/src/datagen/schema.sql +``` + +Open schema.sql — notice every table has a primary key (Datastream's merge mode needs them) and there are deliberately no foreign keys. + +Now the bulk load. This is a **server-side import from Cloud Storage** — the instance pulls the CSVs itself; nothing flows through your tunnel: + +```bash +SQL_SA=$(gcloud sql instances describe cymbal-oltp --format="value(serviceAccountEmailAddress)") +gcloud storage buckets add-iam-policy-binding gs://{{ PROJECT_ID }}-bucket \ + --member=serviceAccount:$SQL_SA --role=roles/storage.objectAdmin +for t in customers products orders order_items payments reviews; do + echo "Importing $t ..." + gcloud sql import csv cymbal-oltp gs://{{ PROJECT_ID }}-bucket/seed/${t}.csv \ + --database=cymbal --table=cymbal.${t} --quiet +done +``` + +This takes about two to three minutes for all six tables. While it runs, read the next section — but don't execute it yet. + +### Prepare logical replication + +Datastream reads Postgres' write-ahead log through a **publication** and a **replication slot**, as a dedicated replication user. The publication defines *which* tables' changes are exposed (`FOR ALL TABLES` here — the safe default), and the slot tracks *how far* a consumer has read: Postgres retains WAL until the slot has consumed it, which is why Datastream never misses a change, even across restarts. (Details: [Configure a Cloud SQL for PostgreSQL source](https://docs.cloud.google.com/datastream/docs/configure-cloudsql-psql).) + +Have a look at replication_setup.sql, then (once the import finished) run it: + +```bash +PGPASSWORD="{{ BK_DB_PASSWORD }}" psql -h localhost -p 5432 -U postgres -d cymbal \ + -v ds_password="{{ BK_DS_PASSWORD }}" \ + -f content/agenticdata/src/datastream/replication_setup.sql +``` + +### Create the bronze dataset + +Both paths below land the CDC data here, so create the destination dataset first: + +```bash +bq mk --location=US --dataset {{ PROJECT_ID }}:cymbal_bronze +``` + +### Provision Datastream: pick your path + +Now you create two **connection profiles** (a PostgreSQL source and a BigQuery destination) and the **CDC stream**. There are two equivalent ways — **pick one**, then continue at *Watch it flow*: + +- **Path A — Console**: click through the Datastream UI, guided by spotlights. Good if you like seeing the wizard and every option. **Unsure? Take this one.** +- **Path B — Command line**: `gcloud`, with agy authoring the stream config. Faster and scriptable. + +*** + +#### Path A — Console (UI) + +**Create the source connection profile.** + +1. Open [Datastream → Connection profiles](https://console.cloud.google.com/datastream/connection-profiles) and click Create profile. +2. Choose the PostgreSQL profile type. +3. Fill in: + - Connection profile name: `cymbal-postgres-profile` + - Region: `{{ REGION }}` + - Hostname or IP: `10.10.0.5` (the PSC endpoint — Datastream private connections don't resolve DNS, so use the IP) + - Port: `5432`, Username: `datastream_user`, Password: `{{ BK_DS_PASSWORD }}`, Database: `cymbal` +4. Click Continue, choose **Private connectivity** and select `cymbal-psc`, then Create. + +**Create the destination connection profile.** + +5. Back on Connection profiles, click Create profile again and choose BigQuery. +6. Name it `cymbal-bq-profile`, region `{{ REGION }}`, then Create. + +**Create and start the stream.** + +7. Go to [Datastream → Streams](https://console.cloud.google.com/datastream/streams) and click Create stream. +8. Stream name `cymbal-cdc-stream`; source type **PostgreSQL**, destination type **BigQuery**. Work through the wizard with Continue: + - **Source**: select `cymbal-postgres-profile`. Run the connectivity test. + - **Configure source**: include the `cymbal` schema (all tables); set **Publication** to `cymbal_pub` and **Replication slot** to `cymbal_slot`. + - **Destination**: select `cymbal-bq-profile`. + - **Configure destination**: choose *Single dataset for all schemas* → `cymbal_bronze`; set the **staleness limit** to `0` seconds. +9. Run the validation, then click Create, and finally Create & start (or open the stream and press **Start**) with a full backfill. + +Skip Path B and continue at **Watch it flow**. + +*** + +#### Path B — Command line (gcloud) + +Two profiles: where the data comes from, and where it goes. Note the hostname — the PSC endpoint IP, because Datastream private connections do not resolve DNS: + +```bash +gcloud datastream connection-profiles create cymbal-postgres-profile --location={{ REGION }} \ + --type=postgresql --display-name=cymbal-postgres-profile \ + --postgresql-hostname=10.10.0.5 --postgresql-port=5432 \ + --postgresql-username=datastream_user --postgresql-password="{{ BK_DS_PASSWORD }}" \ + --postgresql-database=cymbal --private-connection=cymbal-psc +``` + +```bash +gcloud datastream connection-profiles create cymbal-bq-profile --location={{ REGION }} \ + --type=bigquery --display-name=cymbal-bq-profile +``` + +The stream itself is defined by two JSON files that ship with the repository — you work on them directly in `content/agenticdata/src/datastream/` (the shipped destination config still contains a placeholder). Writing API-correct configuration is authoring work — agy's job. In your agy session (started from `~/bootkon`), run: + +``` +/goal Rewrite content/agenticdata/src/datastream/source_config.json and content/agenticdata/src/datastream/destination_config.json for `gcloud datastream streams create`. The PostgreSQL source uses publication "cymbal_pub", replication slot "cymbal_slot", and should include all tables of the "cymbal" schema. The BigQuery destination writes every table into the single dataset "{{ PROJECT_ID }}:cymbal_bronze" with a data freshness of 0 seconds. Use the exact JSON field names of the Datastream v1 API (publication, replicationSlot, includeObjects/postgresqlSchemas, singleTargetDataset/datasetId, dataFreshness). +``` + +**Review what agy wrote**: open source_config.json and destination_config.json. If agy went off-script, git has your back — restore the shipped files and fill in the placeholder yourself: + +```bash +git -C ~/bootkon restore content/agenticdata/src/datastream/ +sed -i "s/PROJECT_ID_PLACEHOLDER/{{ PROJECT_ID }}/" content/agenticdata/src/datastream/destination_config.json +``` + +Create the stream (with a full backfill of the seed data), then flip it to `RUNNING`: + +```bash +gcloud datastream streams create cymbal-cdc-stream --location={{ REGION }} \ + --display-name=cymbal-cdc-stream \ + --source=cymbal-postgres-profile --postgresql-source-config=content/agenticdata/src/datastream/source_config.json \ + --destination=cymbal-bq-profile --bigquery-destination-config=content/agenticdata/src/datastream/destination_config.json \ + --backfill-all +``` + +```bash +gcloud datastream streams update cymbal-cdc-stream --location={{ REGION }} \ + --state=RUNNING --update-mask=state +``` + +*** + +Whichever path you took, Datastream now validates everything (logical decoding, slot, publication, permissions, connectivity) and starts the backfill. The stream reaches **Running** after about two minutes, and the seed data lands in BigQuery roughly a minute later. + +Two design choices in the destination config are worth understanding: the stream runs in **merge mode**, meaning every change event is upserted into the BigQuery table (via the Storage Write API's CDC support), so bronze always mirrors the *current state* of Postgres — the alternative, *append-only*, would keep every event as its own row, giving you a full change history instead. And `dataFreshness: "0s"` tells BigQuery to apply pending changes at query time rather than on a schedule — that's what makes the live demo below feel instant. + +Learn more: +- [BigQuery as a destination (merge vs. append-only)](https://docs.cloud.google.com/datastream/docs/destination-bigquery) +- [BigQuery change data capture](https://docs.cloud.google.com/bigquery/docs/change-data-capture) + +### Watch it flow + +Open [Datastream](https://console.cloud.google.com/datastream/streams) and click cymbal-cdc-stream. Explore the Objects tab — you can watch the per-table backfill progress live. Within a couple of minutes the stream shows **Running** and the six `cymbal_*` tables appear. + +Now make it *live*. Open a **third terminal tab**, and start the activity simulator — Cymbal's customers waking up: + +```bash +cd ~/bootkon +python3 content/agenticdata/src/datagen/simulate.py +``` + +Leave it running. Go to [BigQuery](https://console.cloud.google.com/bigquery), expand {{ PROJECT_ID }} → `cymbal_bronze`, and run this query a few times (use the editor, **not** table preview — preview lags behind CDC): + +```sql +SELECT MAX(order_id) AS latest_order, COUNT(*) AS total_orders +FROM `{{ PROJECT_ID }}.cymbal_bronze.cymbal_orders` +``` + +Watch `latest_order` climb as the simulator inserts. Then prove UPDATEs work end-to-end — back in terminal 1: + +```bash +PGPASSWORD="{{ BK_DB_PASSWORD }}" psql -h localhost -p 5432 -U postgres -d cymbal \ + -c "UPDATE cymbal.customers SET country = 'Iceland', updated_at = now() WHERE customer_id = 42;" +``` + +And in BigQuery (give it a minute or two): + +```sql +SELECT customer_id, full_name, country +FROM `{{ PROJECT_ID }}.cymbal_bronze.cymbal_customers` +WHERE customer_id = 42 +``` + +Customer 42 just moved to Iceland — from a Postgres UPDATE to a BigQuery row, hands-free. That's merge-mode CDC. + +### Challenge: dig deeper + +**\[TASK\]** Take up to 10 minutes: + +1. DELETE an order in psql (pick a high `order_id` from the simulator's output, remove its `order_items` and `payments` rows first) and confirm it disappears from `cymbal_bronze.cymbal_orders`. +2. Every bronze table has an extra `datastream_metadata` column — inspect it and ask agy what `source_timestamp` and `uuid` are for. +3. Ask agy: *"What is the difference between Datastream's merge and append-only modes for BigQuery, and when would I choose each?"* + +### Success + +🎉 Outstanding{% if MY_NAME %}, {{ MY_NAME }}{% endif %}! You built a private, production-style CDC pipeline: a PSC-only Postgres instance, an identity-based tunnel, logical replication, and a running Datastream that mirrors every change into BigQuery in near real time. The bronze layer is alive — time to refine it. 🥉→🥈 diff --git a/content/agenticdata/labs/3_dataform.md b/content/agenticdata/labs/3_dataform.md new file mode 100644 index 0000000..13cd07c --- /dev/null +++ b/content/agenticdata/labs/3_dataform.md @@ -0,0 +1,120 @@ +## Lab 3: The Medallion — Dataform, Authored by agy + + +{{ author('Fabian Hirschmann', 'https://linkedin.com/in/fhirschmann') }} + + + +Your bronze layer is a faithful replica — including every flaw of the source system: duplicate customers, a `shiped` typo, lowercase currencies, orphaned line items, negative payments. In this lab you build the **silver** (cleaned, tested) and **gold** (business marts) layers with **Dataform**. + +The twist: **agy writes all of the SQL**. You direct, review, run, and verify. This is the lab where "agent writes the pipeline" stops being a slide and becomes your terminal. + +We use the open-source **Dataform CLI** — code-first, version-controllable, and a perfect fit for an agentic workflow (compile errors go straight back into agy). In production you would run the same project on a schedule in managed Dataform. + +### About Dataform and the medallion pattern + +Dataform manages SQL transformations as *code*: each table is a **SQLX** file (SQL plus a small config header), dependencies are declared with `${ref(...)}` instead of hardcoded table names, and from those references Dataform compiles a dependency graph and executes it in the right order against BigQuery. **Assertions** are data tests that run with every execution — uniqueness, non-null, arbitrary row conditions — and fail the run when the data breaks a promise; **tags** let you execute just a slice of the graph (you will run `silver` and `gold` separately). + +The **medallion pattern** you are about to build is the standard way to organize such a warehouse: *bronze* holds raw, untouched source data (your CDC replica), *silver* is cleaned and tested, and *gold* holds the business-level marts that people — and, from Lab 5 on, agents — actually consume. Each layer has one job, and problems are fixed at the earliest layer that can see them. + +Learn more: +- [Dataform overview](https://docs.cloud.google.com/dataform/docs/overview) +- [Dataform core concepts](https://docs.cloud.google.com/dataform/docs/dataform-core) +- [The Dataform CLI](https://docs.cloud.google.com/dataform/docs/use-dataform-cli) +- [Assertions](https://docs.cloud.google.com/dataform/docs/assertions) + +### Set up the project + +Install the CLI (takes about a minute): + +```bash +npm install -g @dataform/cli@3 +``` + +The starter project ships with the repository — you work on it directly in `content/agenticdata/src/dataform` (settings + bronze source declarations only; the models are agy's job). **This folder is your home for the rest of this lab** — change into it once, point the project at yours, and set up credentials: + +```bash +cd ~/bootkon/content/agenticdata/src/dataform +sed -i "s/PROJECT_ID_PLACEHOLDER/{{ PROJECT_ID }}/" workflow_settings.yaml +echo "{\"projectId\": \"{{ PROJECT_ID }}\", \"location\": \"US\"}" > .df-credentials.json +``` + +The credentials file tells the CLI to use your logged-in identity (Application Default Credentials) against BigQuery — no keys involved. Check that the `sed` worked by opening workflow_settings.yaml — `defaultProject` should now be `{{ PROJECT_ID }}`. Then have a look at sources.js: it declares the six bronze tables so every model can reference them with `${ref(...)}`, which is what builds the dependency graph — and, later, your lineage. + +### Brief your co-engineer + +The full specification of what to build — source schemas, the known data problems, the target silver/gold tables, and the assertion rules — lives in AGENTS.md, right inside the project. Read it: it's the brief you are about to hand your co-engineer, and keeping it in a file (instead of a giant prompt) makes agy's work reproducible. + +Start agy (you are already in the project folder): + +```bash +agy +``` + +And hand over the brief: + +``` +/goal Read AGENTS.md and build the complete silver and gold layers it specifies. Run `dataform compile` yourself and fix any errors until the project compiles cleanly. +``` + +Watch agy work: it will create the `.sqlx` files, run `dataform compile`, read the errors, and fix its own code. Review the result with `/diff` before accepting. + +Note: **If agy took a wrong turn** or you want to compare with a known-good implementation, the reference lives right next door in `src/dataform_reference/` — for example stg_customers.sqlx and fct_daily_revenue.sqlx. Restore with: + +```bash +cp ~/bootkon/content/agenticdata/src/dataform_reference/definitions/*.sqlx ~/bootkon/content/agenticdata/src/dataform/definitions/ +``` + +### Compile and run + +Exit agy and verify the compilation yourself — trust, but verify: + +```bash +dataform compile +``` + +You should see the models, their dependencies, and the assertions. Now build silver, then gold — each run finishes in well under a minute on the seed data: + +```bash +dataform run --tags silver +``` + +```bash +dataform run --tags gold +``` + +Each run also executes the **assertions** — data tests that fail the run if the cleaning didn't work. If a run fails, paste the error into agy and let it repair its own pipeline; that feedback loop is the whole point of this lab. + +### Verify in the console + +Open [BigQuery](https://console.cloud.google.com/bigquery) and expand your project: + +1. You now have three medallion datasets: `cymbal_bronze`, cymbal_silver and cymbal_gold (plus `cymbal_assertions` — the test results). +2. Click `stg_customers` and check the row count in Details — fewer rows than bronze `cymbal_customers`: the duplicates and invalid emails are gone. +3. Query the flaw you'll never see again: + +```sql +SELECT 'bronze' AS layer, COUNT(*) AS shiped_typos +FROM `{{ PROJECT_ID }}.cymbal_bronze.cymbal_orders` WHERE status = 'shiped' +UNION ALL +SELECT 'silver', COUNT(*) +FROM `{{ PROJECT_ID }}.cymbal_silver.stg_orders` WHERE status = 'shiped' +``` + +4. The payoff: open `fct_daily_revenue` and click the Lineage tab. The bronze→silver→gold graph you see was **not configured by anyone** — BigQuery reports [data lineage](https://docs.cloud.google.com/dataplex/docs/about-data-lineage) automatically from the jobs the Dataform CLI just ran, following the `${ref(...)}` dependencies agy wrote. + +One more thing: your simulator (terminal 3) is still writing to Postgres, and Datastream keeps updating bronze. Re-run `dataform run` at any time and the whole medallion refreshes with the latest data. + +### Challenge: harden the pipeline + +**\[TASK\]** Take up to 10 minutes — pick at least one: + +1. **Incremental gold**: ask agy to convert `fct_daily_revenue` into an incremental table that only processes new days. (Reference prompt in prompts.md.) +2. **Detect, don't just filter**: silver silently drops negative payments. Add a standalone assertion on the **bronze** payments table so the pipeline *alerts* on them — run it with `dataform run --tags challenge` and watch it fail on purpose (982 planted rows). (Reference: assert_bronze_payments_non_negative.sqlx.) +3. **Bonus**: ask agy to add a gold table that classifies `cymbal_reviews` sentiment with BigQuery's `AI.GENERATE` — and discuss with your table what that costs at scale. + +Note: If you are stuck and cannot figure out how to proceed after a few minutes, ask your team captain. + +### Success + +🎉 Bravo{% if MY_NAME %}, {{ MY_NAME }}{% endif %}! You briefed an AI agent, and it wrote, compiled, and repaired a complete medallion pipeline — which *you* reviewed, executed, and verified, assertions and all. Duplicates deduplicated, typos untypo'd, orphans re-homed (well, evicted). The gold layer shines — now let's govern it. 🥇 diff --git a/content/agenticdata/labs/4_governance.md b/content/agenticdata/labs/4_governance.md new file mode 100644 index 0000000..f81a589 --- /dev/null +++ b/content/agenticdata/labs/4_governance.md @@ -0,0 +1,133 @@ +## Lab 4: Governance with Knowledge Catalog + + +{{ author('Fabian Hirschmann', 'https://linkedin.com/in/fhirschmann') }} + + + +You have a working medallion — now make it *trustworthy*. In this lab you govern it with **Knowledge Catalog** (formerly Dataplex Universal Catalog): label the tiers so everyone can find the right data, measure quality automatically, lock down PII, and give business terms a home. In Lab 5 an AI agent becomes a consumer of exactly this governed layer — governance is what keeps agents grounded. + +This lab is console-first: you built everything with code so far; now see how the platform describes itself. + +### About Knowledge Catalog + +Knowledge Catalog (formerly Dataplex Universal Catalog) is Google Cloud's metadata and governance layer. Every dataset, table and column is an **entry** in the catalog; you enrich entries with **aspects** (structured, typed metadata — like the data tiers you are about to define), attach **business glossary** terms, and let **data profiling** and **auto data quality** scans measure the actual content on a schedule or on demand. Because BigQuery reports **lineage** automatically, the catalog also knows where every table came from. All of this metadata is searchable — by humans and, increasingly, by Gemini and agents. That is the deeper point of this lab: a well-curated catalog is the grounding layer that keeps AI answers trustworthy. + +Learn more: +- [Knowledge Catalog overview](https://docs.cloud.google.com/dataplex/docs/introduction) +- [Aspects and aspect types](https://docs.cloud.google.com/dataplex/docs/enrich-entries-metadata) +- [Auto data quality](https://docs.cloud.google.com/dataplex/docs/auto-data-quality-overview) and [data profiling](https://docs.cloud.google.com/dataplex/docs/data-profiling-overview) +- [Column-level security with policy tags](https://docs.cloud.google.com/bigquery/docs/column-level-security-intro) +- [Business glossary](https://docs.cloud.google.com/dataplex/docs/manage-glossaries) +- [Data lineage](https://docs.cloud.google.com/dataplex/docs/about-data-lineage) + +### Label the medallion tiers with aspects + +Aspects are structured metadata attached to catalog entries. We'll create a **Data tier** aspect type and stamp bronze/silver/gold onto the datasets. + +1. Open [Knowledge Catalog](https://console.cloud.google.com/dataplex) and go to Aspect types (under *Metadata types*). +2. Click Create aspect type and use: + - Aspect type ID: `data-tier` + - Display name: `Data tier` + - Location: `global` — this matters: an aspect type can only be attached to entries in the same location or in `global`, and your BigQuery datasets live in the multi-region `us`. A regional aspect type (e.g. `us-central1`) will *not* show up when you try to attach it below. +3. Add a field (under *Template*, click **Add Field**): + - Type: **Enum**, ID: `tier`, Display name: `Tier` + - Enum values (**Add an Enum Value** three times): `bronze`, `silver`, `gold` + - Mark it **required**. +4. Click Create. + +Now attach it. Go to Search, search for `cymbal_gold`, and open the dataset entry: + +5. In the entry's details, find Aspects → under *Optional aspects* click *Add*, filter for **Data tier**, set Tier to `gold` → save. +6. Repeat for `cymbal_silver` (`silver`) and `cymbal_bronze` (`bronze`). + +Verify the point of the exercise: in the catalog search bar, filter by your new aspect (e.g. search for `cymbal` and use the aspect filter for `Data tier = gold`) — anyone in the company can now find the *consumable* data without asking around. + +### Measure quality automatically + +In Lab 3 your assertions tested what Dataform *built*. Knowledge Catalog's **auto data quality** watches tables *continuously* — no pipeline required. Let's point it at the flaws you know are in bronze. + +First, allow the Dataplex service agent to use your scan service account: + +```bash +gcloud iam service-accounts add-iam-policy-binding dataquality-service-account@{{ PROJECT_ID }}.iam.gserviceaccount.com \ + --member=serviceAccount:service-{{ PROJECT_NUMBER }}@gcp-sa-dataplex.iam.gserviceaccount.com \ + --role=roles/iam.serviceAccountTokenCreator +``` + +Now profile the customer data: + +1. In Knowledge Catalog, open Data profiling & quality. +2. Click Create data profile scan: + - Display name: `cymbal-profile-bronze-customers` + - Table: browse to `cymbal_bronze` → `cymbal_customers` + - Scope *Entire data*, sampling *All data*, publish results on + - Credential type: **Service account** → `dataquality-service-account` + - Schedule: On-demand +3. Create it, open it, and click Run now. +4. When the job finishes (a few minutes), explore the results — look at the `country` column: there's your planted ~1.5% NULL rate, and the `email` column's distinct count hints at the duplicates. + +Then hold bronze orders to a standard: + +5. Back on the same page, click Create data quality scan: + - Display name: `cymbal-dq-bronze-orders` + - Table: `cymbal_bronze` → `cymbal_orders` + - Credential type: Service account → `dataquality-service-account`, On-demand +6. Add two rules (rule type *Row check* / validity): + - `status IN ('pending','paid','shipped','delivered','cancelled','returned')` — dimension *Validity* + - `order_ts <= CURRENT_TIMESTAMP()` — dimension *Accuracy* +7. Run the scan. **It fails — on purpose.** The `shiped` typo and the future-dated orders you saw in Lab 3 are now caught by governance, not just by your pipeline. Discuss with your table: the same rules would pass on `cymbal_silver.stg_orders` — why keep both layers scanned? (If you have time, clone the scan onto silver and prove it passes.) + +### Lock down PII + +`stg_customers.email` is personal data. Enforce column-level security with a policy tag: + +1. Open [BigQuery policy tags](https://console.cloud.google.com/bigquery/policy-tags) and click Create taxonomy: + - Taxonomy name: `cymbal-governance`, location `us` + - Policy tag: `PII`, description: `Personal data — restricted` +2. Create it, then toggle Enforce access control on. +3. In [BigQuery](https://console.cloud.google.com/bigquery), open `cymbal_silver` → `stg_customers` → Edit schema, select the `email` column, click *Add policy tag*, and pick `cymbal-governance > PII`. Save. + +Now prove it works — this query **must fail** with an access-denied error on the tagged column: + +```sql +SELECT email FROM `{{ PROJECT_ID }}.cymbal_silver.stg_customers` LIMIT 5 +``` + +And this one works fine: + +```sql +SELECT * EXCEPT (email) FROM `{{ PROJECT_ID }}.cymbal_silver.stg_customers` LIMIT 5 +``` + +You are the project owner and *still* can't read that column — fine-grained access is a separate grant (Fine-Grained Reader). That's exactly the guarantee you want before letting AI agents loose on the warehouse. (At scale you wouldn't tag by hand: [Sensitive Data Protection discovery](https://docs.cloud.google.com/sensitive-data-protection/docs/data-profiles) profiles your tables continuously and pushes its findings into the catalog as aspects.) + +### Give the business a vocabulary + +1. In Knowledge Catalog, open Glossaries and create glossary `Cymbal Business Glossary` (location `us-central1`). +2. Add a term: **Lifetime value** — *"Gross revenue of a customer's non-cancelled orders, in the order currency. Source of truth: cymbal_gold.dim_customer_360.lifetime_value."* +3. Open the `dim_customer_360` entry via catalog Search, go to its schema, select the `lifetime_value` column and attach the term. + +Then try the agentic side of governance — in the catalog Search, ask in natural language: + +``` +Which tables contain revenue by day? +``` + +Gemini-powered search reads the same metadata you just curated — every aspect, term, and description you add makes both humans *and* agents smarter. + +### Admire the lineage + +One more look: open `cymbal_gold.fct_daily_revenue` in BigQuery and its Lineage tab. Bronze→silver→gold, captured automatically from the Dataform runs. (Datastream's Postgres→bronze hop publishes its metadata to the catalog in Preview, but doesn't draw lineage edges yet — watch that space.) + +### Challenge: publish a data product + +**\[TASK\]** Take up to 10 minutes: package `cymbal_gold` as a **data product** — the contract your data agent will consume in Lab 5. + +- In Knowledge Catalog, find Data products and create one from the three gold tables. +- Let Gemini generate the product documentation — then *edit it*: you are accountable for the contract, the model just drafts it. +- Explore what a consumer would see before requesting access. + +### Success + +🎉 Splendid{% if MY_NAME %}, {{ MY_NAME }}{% endif %}! Your platform now explains itself: tiers are labeled, quality is measured continuously (and honestly — bronze fails, as it should), PII is locked down even against project owners, business terms live next to the data, and lineage draws itself. Governance done — the agents can come. 🛡️ diff --git a/content/agenticdata/labs/5_data_agent.md b/content/agenticdata/labs/5_data_agent.md new file mode 100644 index 0000000..c234ad6 --- /dev/null +++ b/content/agenticdata/labs/5_data_agent.md @@ -0,0 +1,90 @@ +## Lab 5: The BigQuery Data Agent + + +{{ author('Fabian Hirschmann', 'https://linkedin.com/in/fhirschmann') }} + + + +Time for the first agent. **Conversational analytics data agents** live inside BigQuery: you point one at your governed gold tables, give it context, and business users chat with the data in natural language — the agent writes and runs the SQL. In Lab 6 this same agent becomes a *service* that other agents call over A2A, so make it good. + +Note the payoff of your work so far: the agent will only ever see the **gold** layer — cleaned by your silver models, quality-scanned, PII-tagged, and documented. Grounded agents start with governed data. + +### About data agents + +A conversational analytics **data agent** bundles three things: the *knowledge sources* it may query (your gold tables), *context* that teaches it your business language (the system instructions you are about to write), and *verified queries* — blessed question/SQL pairs it prefers over improvising. When a user asks a question, Gemini translates it into SQL against exactly those sources, runs it as a regular BigQuery job — so IAM and your column-level security from Lab 4 fully apply — and summarizes the result, showing its SQL as it goes. And because the underlying **Conversational Analytics API** is GA, a *published* agent is not just a console toy: it is a callable resource with its own IAM — which is precisely what Lab 6 exploits. + +Learn more: +- [Create and use data agents in BigQuery](https://docs.cloud.google.com/bigquery/docs/create-data-agents) +- [Conversational Analytics API](https://docs.cloud.google.com/gemini/data-agents/conversational-analytics-api/overview) + +### Draft the instructions with agy + +A data agent is only as good as its context. That context is derived from your schemas — authoring work, so it goes to agy. Run this in a terminal (non-interactive mode; the prompt is also in prompts.md): + +```bash +agy -p "Read the table schemas: bq show --schema {{ PROJECT_ID }}:cymbal_gold.fct_daily_revenue ; bq show --schema {{ PROJECT_ID }}:cymbal_gold.dim_customer_360 ; bq show --schema {{ PROJECT_ID }}:cymbal_gold.fct_product_performance. Then draft system instructions for a BigQuery conversational data agent over these three tables: synonyms business users might use (revenue, sales, LTV, best sellers), which table answers which kind of question, default groupings (order_date, currency, category), and columns to exclude from summaries (email). Output only the instructions text." +``` + +Review the output — you are about to make it the agent's world-view. A solid baseline, if you prefer to use it directly or to compare: + +``` +Tables and their roles: +- fct_daily_revenue: revenue, sales, turnover questions. Group by order_date and currency by default. gross_revenue is SUM(qty * unit_price) excluding cancelled orders. +- dim_customer_360: customer questions. "LTV", "lifetime value", "best customers" -> lifetime_value. One row per deduplicated customer. +- fct_product_performance: product questions. "best sellers" -> units_sold; "most profitable" -> gross_margin. Group by category for overviews. + +Rules: +- Amounts are per currency; never sum across currencies without saying so. +- Exclude the email column from any output or summary. +- "Orders" means non-cancelled orders unless the user asks otherwise. +- Data freshness: near real time via CDC + Dataform runs; mention the latest order_date when answering trend questions. +``` + +### Create the agent + +1. Go to the [BigQuery Console](https://console.cloud.google.com/bigquery). +2. Expand {{ PROJECT_ID }} and click Agents. +3. Click Create agent. +4. Name it ``cymbal-data-agent`` — Lab 6 refers to it by exactly this name — and add a description like *"Analytics over Cymbal's gold layer"*. +5. Click *Add Source* and select the three gold tables: ``fct_daily_revenue``, ``dim_customer_360``, ``fct_product_performance``. +6. Paste your instructions (agy's draft, reviewed, or the baseline above) into the instructions box. + +### Add a verified query + +Verified queries are blessed SQL the agent prefers over improvising — your accuracy anchor. Add one for the most common question, total revenue by day: + +```sql +SELECT order_date, currency, SUM(gross_revenue) AS revenue +FROM `{{ PROJECT_ID }}.cymbal_gold.fct_daily_revenue` +GROUP BY order_date, currency +ORDER BY order_date DESC +``` + +with the natural-language question: ``What was our revenue by day?`` + +### Publish and converse + +In the top bar: +1. Click ``Save`` +2. Then ``Publish`` — publishing is what makes the agent callable via the Conversational Analytics API, which Lab 6 depends on. + +Now click ``Create conversation`` and interrogate your platform. Try these, then improvise: + +- What was our total revenue in EUR in the last 30 days? +- Who are our top 10 customers by lifetime value, and which countries are they from? +- Which product category has the best gross margin? +- Are there any days with unusually high revenue? What might explain them? + +Watch what comes back: the agent shows the SQL it generated against your gold tables. You spent three labs making those tables mean something — this is where it pays off. (Also try asking for customer *emails* — the agent queries gold, where governance applies.) + +### Challenge: stress-test the grounding + +**\[TASK\]** Take up to 5 minutes: try to make the agent say something wrong. + +- Ask about data it doesn't have ("What's our warehouse inventory?") — does it admit ignorance or hallucinate? +- Ask an ambiguous question ("How are sales?") — does it apply your default groupings from the instructions? +- Refine the instructions based on what you find and re-publish. Iterating on agent context *is* data work now. + +### Success + +🎉 Excellent{% if MY_NAME %}, {{ MY_NAME }}{% endif %}! Cymbal now has a published data agent: natural language in, governed SQL out, grounded in the gold layer you built and documented. One agent down — in the finale, it gets a colleague. 🤖 diff --git a/content/agenticdata/labs/6_a2a.md b/content/agenticdata/labs/6_a2a.md new file mode 100644 index 0000000..6fda64c --- /dev/null +++ b/content/agenticdata/labs/6_a2a.md @@ -0,0 +1,116 @@ +## Lab 6: The A2A Finale — Agents Talking to Agents + + +{{ author('Fabian Hirschmann', 'https://linkedin.com/in/fhirschmann') }} + + + +The finale. You'll build **two agents** with the **Agent Development Kit (ADK)** and connect them over the **A2A protocol** — the open standard (now at the Linux Foundation) for agents calling other agents across process and organization boundaries: + +- ``cymbal_analyst`` — wraps the BigQuery data agent from Lab 5 and is **exposed as an A2A server**. The *analytical plane*. +- ``cymbal_concierge`` — the agent you chat with. It **delegates analytics to the analyst via A2A** and has one tool of its own: a live order lookup against Postgres through your IAP tunnel. The *operational plane*. + +One conversation, two planes, and the entire platform you built today underneath. + +### About ADK and the A2A protocol + +The **Agent Development Kit (ADK)** is Google's open-source framework for building agents in code: an `Agent` combines a model, an instruction, **tools** (plain Python functions the model may call — their docstrings become the tool documentation the model reasons over) and optionally **sub-agents** it can delegate whole conversations to. The **A2A protocol** (agent-to-agent, governed by the Linux Foundation) standardizes how agents call *each other* across process, vendor and framework boundaries: every A2A server publishes an **agent card** — a JSON discovery document describing its skills and endpoint — and clients like ADK's `RemoteA2aAgent` read that card to route work to it. That is the point of this finale: once exposed over A2A, your analyst could be consumed by *any* A2A-speaking agent in the company, written in any framework — not just by your concierge. + +Learn more: +- [Agent Development Kit](https://adk.dev/) +- [A2A protocol specification](https://a2a-protocol.org/) +- [Exposing an ADK agent over A2A](https://adk.dev/a2a/quickstart-exposing/) + +First, verify the foundation: open [BigQuery](https://console.cloud.google.com/bigquery) → Agents and confirm ``cymbal-data-agent`` shows as **published**. That's the resource the analyst will call. + +### Set up the project + +The agent project ships with the repository — you work directly in `content/agenticdata/src/adk`. There is no configuration file to create: everything the agents need (`GOOGLE_CLOUD_PROJECT`, the pinned Gemini model, the data agent ID, the database password) was exported to your `~/.bashrc` by `bk-bootstrap` in Lab 1, and the agents read it from the environment. They authenticate to Gemini via Vertex AI with your logged-in credentials — no API keys anywhere. + +All Python dependencies were installed by the bootstrap too, so just confirm the ADK CLI is there and check your configuration: + +```bash +cd ~/bootkon/content/agenticdata/src/adk +adk --version +env | grep -E 'BK_|GOOGLE_CLOUD' | grep -v PASSWORD +``` + +### Build the analyst with agy + +The analyst is authoring work — brief your co-engineer. The repository ships a finished analyst as reference, so for the authentic experience, clear the stage first: + +```bash +rm -rf cymbal_analyst +``` + +The analyst's full specification — its one tool, the Conversational Analytics wiring, the A2A server — lives in AGENTS.md in the project folder. Read it first, then start ``agy`` (you are already in ``content/agenticdata/src/adk``) and paste: + +``` +/goal Read AGENTS.md, then create the files of the cymbal_analyst package (cymbal_analyst/__init__.py, agent.py, a2a_server.py) exactly as it describes. +``` + +Review with `/diff`. As always, git is your safety net — the shipped reference is one command away: + +```bash +git -C ~/bootkon restore content/agenticdata/src/adk/cymbal_analyst/ +``` + +After restoring (or for comparison with agy's version), open the reference agent.py and ca_tool.py. + +The concierge is pre-built and already in place (you'll extend it in the challenge) — read its code: note ``RemoteA2aAgent`` (the A2A client side) and the ``check_order_status`` tool hitting ``localhost:5432``. + +### Expose the analyst over A2A + +In your current terminal: + +```bash +cd ~/bootkon/content/agenticdata/src/adk +uvicorn cymbal_analyst.a2a_server:a2a_app --host 127.0.0.1 --port 8001 +``` + +Leave it serving. Open a **new terminal**, `cd ~/bootkon/content/agenticdata/src/adk`, and fetch the analyst's business card: + +```bash +curl -s localhost:8001/.well-known/agent-card.json | python3 -m json.tool +``` + +That JSON is the **A2A agent card** — name, skills, endpoint — the standard discovery document any A2A-speaking agent (from any vendor, any framework) uses to find and call yours. Your data platform just became an interoperable service. *(Spec ≥1.0 serves it at `agent-card.json`; older tutorials say `agent.json` — stale.)* + +### Talk to the concierge + +Quick pre-flight: the concierge needs the IAP tunnel from Lab 2 (terminal 2) — if it died in the break, restart it. Then launch the ADK developer UI: + +```bash +adk web --port 8000 +``` + +Open it via **Web Preview** (the Web Preview button in the Cloud Shell toolbar → *Preview on port 8000*), and select **cymbal_concierge**. + +Now run the two-plane demo: + +1. **Analytical** — ask: ``Which product category made the most revenue in the last 30 days?`` + Watch the event trace in the UI: concierge → *A2A call to cymbal_analyst on :8001* → Conversational Analytics API → SQL on `cymbal_gold`. Four systems, one question. +2. **Operational** — ask: ``What's the status of order 421337?`` + This time: concierge → `check_order_status` → IAP tunnel → live Postgres. If the simulator deleted it, the concierge says so — it's reading production truth, not a copy. (Grab a fresh existing id from the simulator terminal and try again.) +3. **Both at once** — ask: ``Is order 421337 delivered, and how does its product category perform overall?`` and watch it route each half correctly. + +The point to savor: the analyst answers from data that started this afternoon as Postgres rows, flowed through Datastream into bronze, was cleaned by agy's Dataform code into gold, got governed in Knowledge Catalog, and is served by a published data agent — now consumed by another agent, over an open protocol, on behalf of a user. **That is agentic data.** + +### Challenge: production moves + +**\[TASK\]** Pick any, time permitting: + +1. **Extend the concierge**: have agy add a `recent_orders(customer_email)` tool (prompt in prompts.md). +2. **Tool, not teammate**: convert the analyst from a `sub_agents` delegation into an `AgentTool` call and observe how the conversation flow changes. +3. **Built-in toolset**: swap the hand-written CA tool for ADK's `DataAgentToolset` (`google.adk.tools.data_agent`) and compare. +4. **Go remote for real**: deploy the analyst's A2A server to Cloud Run and point `RemoteA2aAgent` at the Cloud Run URL — same protocol, now truly distributed. (The Conversational Analytics API itself has a native A2A endpoint in Preview — ask your team captain about it.) + +Note: If you are stuck and cannot figure out how to proceed after a few minutes, ask your team captain. + +### Success + +🎉 Standing ovation{% if MY_NAME %}, {{ MY_NAME }}{% endif %}! 🚀 + +Count what you built in one afternoon: a private CDC pipeline (Cloud SQL → Datastream → BigQuery), an agent-authored medallion architecture with data tests, catalog governance with enforced PII protection, a published BigQuery data agent — and a multi-agent system where an ADK concierge routes between live operations and governed analytics over the A2A protocol. + +You didn't just use AI on data. You built the platform that lets agents be *trusted* with data. Cymbal would hire you on the spot. 🏆 diff --git a/content/agenticdata/option1-vpc-peering-proxy.diff b/content/agenticdata/option1-vpc-peering-proxy.diff new file mode 100644 index 0000000..27092f0 --- /dev/null +++ b/content/agenticdata/option1-vpc-peering-proxy.diff @@ -0,0 +1,168 @@ +Option 1 fallback: VPC-peering private connectivity + reverse-proxy VM +======================================================================= + +Apply this patch ONLY if the PSC path fails on the target environment (e.g. +the sandbox blocks PSC-enabled Cloud SQL instances or Datastream PSC +interfaces). + +It switches Labs 1 and 2 from "PSC end-to-end" to the classic documented +workaround: Cloud SQL on a private IP via Private Services Access, a +Datastream private connection over VPC peering (/29 subnet 10.20.0.0/29), +and the jump VM promoted to Google's official reverse proxy (VPC peering is +non-transitive, so Datastream cannot reach the Cloud SQL services network, +nor PSC endpoints, directly). Both the console (Path A) and gcloud (Path B) +provisioning flows are patched; the shared jump-VM startup script +(src/jumpvm-startup.sh) is re-targeted at the database's private IP via sed +into /tmp before VM creation. + +Apply from the repository root: + + git apply content/agenticdata/option1-vpc-peering-proxy.diff + +If the labs have drifted since this diff was written: + + git apply --recount -C1 content/agenticdata/option1-vpc-peering-proxy.diff + +--- a/content/agenticdata/labs/1_environment_setup.md 2026-07-03 09:04:24 ++++ b/content/agenticdata/labs/1_environment_setup.md 2026-07-03 09:04:24 +@@ -57,7 +57,7 @@ + + ### Create the network path + +-Our database will have **no public IP**. Instead, Datastream will reach it through Private Service Connect (PSC). First, create a VPC and a subnet: ++Our database will have **no public IP**. Instead, Datastream will reach it through private connectivity (VPC peering) and a reverse-proxy VM. First, create a VPC and a subnet: + + ```bash + gcloud compute networks create cymbal-vpc --subnet-mode=custom +@@ -68,25 +68,27 @@ + --network=cymbal-vpc --region={{ REGION }} --range=10.10.0.0/24 + ``` + +-Datastream connects into your VPC through a **network attachment** — the entry door for its PSC interface: ++The database will live at a private IP inside a Google-managed services network, connected to your VPC via **Private Services Access** (VPC peering). Allocate a range and connect it: + + ```bash +-gcloud compute network-attachments create cymbal-attachment --region={{ REGION }} \ +- --connection-preference=ACCEPT_AUTOMATIC --subnets=cymbal-subnet ++gcloud compute addresses create cymbal-psa --global --purpose=VPC_PEERING \ ++ --prefix-length=16 --network=cymbal-vpc ++gcloud services vpc-peerings connect --service=servicenetworking.googleapis.com \ ++ --ranges=cymbal-psa --network=cymbal-vpc + ``` + +-Note: `ACCEPT_AUTOMATIC` keeps this workshop simple. In production you would use `ACCEPT_MANUAL` with a `--producer-accept-list`, discovering the Datastream tenant project via `gcloud datastream private-connections create --validate-only`. ++Note: VPC Network Peering is not transitive: Datastream's own peered network cannot reach the Cloud SQL services network directly. That is why this variant adds a small reverse-proxy VM in Lab 2 — Google's documented workaround. + +-So what did you just build? **Private Service Connect** is Google Cloud's way of exposing a service across VPC boundaries without peering entire networks or using public IPs: the producer (here: your Cloud SQL instance) publishes a *service attachment*, and consumers plug an *endpoint* into it — traffic stays on Google's backbone, and each side only sees the single socket it was given. The **network attachment** you created is the reverse construct: it lets a Google-managed producer (Datastream) place a network interface *into* your VPC. You will connect both halves in Lab 2. ++So what did you just build? **Private Services Access** connects your VPC to a Google-managed services network over VPC peering — your database will get a private IP inside that peered network. Because peering is not transitive, nothing outside your VPC (including Datastream) can reach that IP directly; Lab 2 adds Google's documented reverse-proxy VM as the bridge. + + Learn more: +-- [Private Service Connect](https://docs.cloud.google.com/vpc/docs/private-service-connect) +-- [Network attachments](https://docs.cloud.google.com/vpc/docs/about-network-attachments) +-- [Cloud SQL and Private Service Connect](https://docs.cloud.google.com/sql/docs/postgres/about-private-service-connect) ++- [Private Services Access](https://docs.cloud.google.com/vpc/docs/private-services-access) ++- [Cloud SQL private IP](https://docs.cloud.google.com/sql/docs/postgres/private-ip) ++- [Datastream private connectivity](https://docs.cloud.google.com/datastream/docs/private-connectivity) + + ### Launch your operational database + +-This is Cymbal's production order database: PostgreSQL on Cloud SQL, with logical decoding enabled at creation time (Datastream needs it for CDC) and Private Service Connect instead of a public IP. The `--async` flag returns immediately — the instance builds in the background for the next 10–15 minutes while you continue: ++This is Cymbal's production order database: PostgreSQL on Cloud SQL, with logical decoding enabled at creation time (Datastream needs it for CDC) and a private IP (Private Services Access) instead of a public IP. The `--async` flag returns immediately — the instance builds in the background for the next 10–15 minutes while you continue: + + {% if ON_ARGOLIS %} + ❗ You are on Argolis. Instance and VM creation in this stream may be blocked by organization policies (for example `constraints/compute.requireOsLogin` or `constraints/compute.requireShieldedVm`). If a create command fails with a policy error, disable the constraint under IAM & Admin → Organization Policies and retry. +@@ -99,8 +101,7 @@ + --region={{ REGION }} \ + --root-password=$BK_DB_PASSWORD \ + --database-flags=cloudsql.logical_decoding=on \ +- --enable-private-service-connect \ +- --allowed-psc-projects={{ PROJECT_ID }} \ ++ --network=projects/{{ PROJECT_ID }}/global/networks/cymbal-vpc \ + --no-assign-ip \ + --async + ``` +@@ -117,7 +118,7 @@ + ```bash + gcloud datastream private-connections create cymbal-psc --location={{ REGION }} \ + --display-name=cymbal-psc \ +- --network-attachment=projects/{{ PROJECT_ID }}/regions/{{ REGION }}/networkAttachments/cymbal-attachment ++ --vpc=cymbal-vpc --subnet=10.20.0.0/29 + ``` + + ### Stage the seed data +--- a/content/agenticdata/labs/2_replication.md 2026-07-03 09:04:24 ++++ b/content/agenticdata/labs/2_replication.md 2026-07-03 09:04:24 +@@ -34,35 +34,36 @@ + + ❗ If this prints `NOT_FOUND` or `FAILED`, the creation lost a race against IAM propagation for the freshly created Datastream service agent. No harm done — just re-run the `gcloud datastream private-connections create` command from Lab 1 (delete the failed one first if `FAILED`), and it will go through. + +-### Create the Private Service Connect endpoint ++### Look up the database's private IP + +-Your instance exposes a **service attachment** — a private socket other networks can plug into. Create an endpoint for it in your VPC at the reserved IP `10.10.0.5`: ++Your instance lives at a private IP inside the Google-managed services network peered with your VPC. Look it up and store it: + + ```bash +-SA_URI=$(gcloud sql instances describe cymbal-oltp --format="value(pscServiceAttachmentLink)") +-gcloud compute addresses create cymbal-endpoint-ip --region={{ REGION }} \ +- --subnet=cymbal-subnet --addresses=10.10.0.5 +-gcloud compute forwarding-rules create cymbal-endpoint --region={{ REGION }} \ +- --address=cymbal-endpoint-ip --network=cymbal-vpc \ +- --target-service-attachment=$SA_URI --allow-psc-global-access ++DB_PRIVATE_IP=$(gcloud sql instances describe cymbal-oltp --format="value(ipAddresses[0].ipAddress)") ++echo "export DB_PRIVATE_IP=$DB_PRIVATE_IP" >> ~/.bashrc ++echo "Cymbal database private IP: $DB_PRIVATE_IP" + ``` + +-From now on, `10.10.0.5` **is** your database — for Datastream and for the jump VM below. ++From now on, the **reverse-proxy VM below** is how both Datastream and you reach the database — VPC peering is not transitive, so nothing outside your VPC can reach that IP directly. + +-### Create the jump VM ++### Create the reverse-proxy VM + +-Cloud Shell lives outside your VPC and a PSC-only instance has no public IP, so you need a tiny helper: an e2-micro VM (no external IP either!) that forwards port 5432 to the database endpoint. Its startup script is four lines of iptables. You will reach the VM through an **IAP tunnel** — identity-based, no IPs exposed anywhere: ++Cloud Shell lives outside your VPC, and Datastream's peered network cannot reach the database's services network either. One tiny VM solves both: an e2-micro reverse proxy (no external IP!) forwarding port 5432 to the database's private IP — Google's documented pattern for Datastream with Cloud SQL private IP. Substitute your database IP into the startup script, then create the VM: + + ```bash ++sed "s/10.10.0.5/$DB_PRIVATE_IP/" content/agenticdata/src/jumpvm-startup.sh > /tmp/jumpvm-startup.sh + gcloud compute instances create cymbal-jump --zone={{ REGION }}-a \ + --machine-type=e2-micro --subnet=cymbal-subnet --no-address --can-ip-forward \ +- --metadata-from-file=startup-script=content/agenticdata/src/jumpvm-startup.sh ++ --metadata-from-file=startup-script=/tmp/jumpvm-startup.sh + ``` + + ```bash + gcloud compute firewall-rules create allow-iap-ingress --network=cymbal-vpc \ + --direction=INGRESS --action=allow --rules=tcp:22,tcp:5432 \ + --source-ranges=35.235.240.0/20 ++gcloud compute firewall-rules create allow-datastream-ingress --network=cymbal-vpc \ ++ --direction=INGRESS --action=allow --rules=tcp:5432 \ ++ --source-ranges=10.20.0.0/29 + ``` + + ### Open the tunnel +@@ -147,7 +148,7 @@ + 3. Fill in: + - Connection profile name: `cymbal-postgres-profile` + - Region: `{{ REGION }}` +- - Hostname or IP: `10.10.0.5` (the PSC endpoint — Datastream private connections don't resolve DNS, so use the IP) ++ - Hostname or IP: your **reverse-proxy VM's internal IP** — run `gcloud compute instances describe cymbal-jump --zone={{ REGION }}-a --format='value(networkInterfaces[0].networkIP)'` to get it (Datastream can't reach the database directly and doesn't resolve DNS) + - Port: `5432`, Username: `datastream_user`, Password: `{{ BK_DS_PASSWORD }}`, Database: `cymbal` + 4. Click Continue, choose **Private connectivity** and select `cymbal-psc`, then Create. + +@@ -172,12 +173,13 @@ + + #### Path B — Command line (gcloud) + +-Two profiles: where the data comes from, and where it goes. Note the hostname — the PSC endpoint IP, because Datastream private connections do not resolve DNS: ++Two profiles: where the data comes from, and where it goes. Note the hostname — the **reverse-proxy VM's internal IP**, because Datastream cannot reach the database directly and does not resolve DNS in private connections: + + ```bash ++VM_IP=$(gcloud compute instances describe cymbal-jump --zone={{ REGION }}-a --format='value(networkInterfaces[0].networkIP)') + gcloud datastream connection-profiles create cymbal-postgres-profile --location={{ REGION }} \ + --type=postgresql --display-name=cymbal-postgres-profile \ +- --postgresql-hostname=10.10.0.5 --postgresql-port=5432 \ ++ --postgresql-hostname=$VM_IP --postgresql-port=5432 \ + --postgresql-username=datastream_user --postgresql-password="{{ BK_DS_PASSWORD }}" \ + --postgresql-database=cymbal --private-connection=cymbal-psc + ``` diff --git a/content/agenticdata/requirements.txt b/content/agenticdata/requirements.txt new file mode 100644 index 0000000..cca6b6d --- /dev/null +++ b/content/agenticdata/requirements.txt @@ -0,0 +1,10 @@ +# Pinned for reproducibility across ~100 identical sandbox environments. +# Re-verify every pin during the smoke test (CONCEPT.md, section 12) -- +# google-adk and google-cloud-geminidataanalytics move fast. +faker==37.4.0 +psycopg[binary]==3.2.13 +google-adk[a2a]==2.3.0 +# google-adk[a2a] does NOT pull the http-server extras that to_a2a() needs +# at runtime (starlette/sse-starlette) -- verified live in the smoke test. +a2a-sdk[http-server]==0.3.26 +google-cloud-geminidataanalytics==0.13.1 diff --git a/content/agenticdata/src/adk/AGENTS.md b/content/agenticdata/src/adk/AGENTS.md new file mode 100644 index 0000000..e3834fc --- /dev/null +++ b/content/agenticdata/src/adk/AGENTS.md @@ -0,0 +1,55 @@ +# Cymbal agents project — agent context + +An ADK project (google-adk 2.x is installed) with two agents: + +## cymbal_analyst/ — the analytics specialist (built in Lab 6) + +- `agent.py` defines + `root_agent = Agent(name="cymbal_analyst", model=..., instruction=..., tools=[...])` + (note: the keyword is `instruction`, singular) with exactly ONE function tool: it sends a natural-language question to the + published BigQuery data agent via the `google-cloud-geminidataanalytics` + `DataChatServiceClient` and returns the streamed text parts joined together. +- The Conversational Analytics API is newer than your training data — do NOT + guess its shape. The stateless chat call is exactly: + + ```python + client = geminidataanalytics.DataChatServiceClient() + message = geminidataanalytics.Message() + message.user_message.text = question + context = geminidataanalytics.DataAgentContext() + context.data_agent = f"projects/{project}/locations/global/dataAgents/{agent_id}" + request = geminidataanalytics.ChatRequest( + parent=f"projects/{project}/locations/global", + messages=[message], + data_agent_context=context, + ) + for reply in client.chat(request=request): + # collect reply.system_message.text.parts when present + ``` +- `a2a_server.py` exposes it over the A2A protocol via + `google.adk.a2a.utils.agent_to_a2a.to_a2a(root_agent, port=8001)` as the + module attribute `a2a_app` (served with uvicorn). +- `__init__.py` must do `from . import agent`. +- Instruction: answer only aggregate analytics questions from the gold layer, + strictly based on the tool result; individual live orders are the + concierge's job. + +## cymbal_concierge/ — the root agent (pre-built, do not rewrite) + +`RemoteA2aAgent` sub-agent for the analyst (agent card on localhost:8001) +plus a `check_order_status` tool that queries live Postgres through the IAP +tunnel on localhost:5432. + +## Conventions + +- Exact imports: `from google.adk.agents import Agent`, + `from google.adk.a2a.utils.agent_to_a2a import to_a2a`, + `from google.cloud import geminidataanalytics`. +- All configuration comes from environment variables: + `GOOGLE_CLOUD_PROJECT`, `GOOGLE_GENAI_USE_VERTEXAI`, `GOOGLE_CLOUD_LOCATION`, + `BK_CYMBAL_MODEL` (default `gemini-2.5-flash`), `BK_DATA_AGENT_ID`, + `BK_CYMBAL_DB_HOST`/`BK_CYMBAL_DB_PORT`, `BK_DB_PASSWORD`. + Never create config files and never hardcode credentials. +- Read the model as `os.environ.get("BK_CYMBAL_MODEL", "gemini-2.5-flash")`. +- Keep tools small and readable; docstrings are the tool documentation the + model reasons over. diff --git a/content/agenticdata/src/adk/cymbal_analyst/__init__.py b/content/agenticdata/src/adk/cymbal_analyst/__init__.py new file mode 100644 index 0000000..e7186e1 --- /dev/null +++ b/content/agenticdata/src/adk/cymbal_analyst/__init__.py @@ -0,0 +1 @@ +from . import agent # noqa: F401 (ADK discovers root_agent via this import) diff --git a/content/agenticdata/src/adk/cymbal_analyst/a2a_server.py b/content/agenticdata/src/adk/cymbal_analyst/a2a_server.py new file mode 100644 index 0000000..2e723c6 --- /dev/null +++ b/content/agenticdata/src/adk/cymbal_analyst/a2a_server.py @@ -0,0 +1,20 @@ +"""Exposes cymbal_analyst as an A2A server. + +Run from the project root (content/agenticdata/src/adk): + + uvicorn cymbal_analyst.a2a_server:a2a_app --host 127.0.0.1 --port 8001 + +Configuration comes from the environment (~/.bashrc, set by bk-bootstrap). + +to_a2a() auto-generates the agent card, served at +http://localhost:8001/.well-known/agent-card.json (A2A spec >= 0.3/1.0 -- +older tutorials still show agent.json, which is stale). +The port passed to to_a2a() is baked into the card and MUST match uvicorn's +--port, or consumers will be pointed at the wrong address. +""" + +from google.adk.a2a.utils.agent_to_a2a import to_a2a + +from .agent import root_agent + +a2a_app = to_a2a(root_agent, port=8001) diff --git a/content/agenticdata/src/adk/cymbal_analyst/agent.py b/content/agenticdata/src/adk/cymbal_analyst/agent.py new file mode 100644 index 0000000..27a3d95 --- /dev/null +++ b/content/agenticdata/src/adk/cymbal_analyst/agent.py @@ -0,0 +1,39 @@ +"""cymbal_analyst -- the analytics specialist agent. + +It answers business questions by consulting Cymbal's published BigQuery data +agent (Conversational Analytics) and is exposed to other agents over the A2A +protocol by a2a_server.py. + +All configuration (GOOGLE_CLOUD_PROJECT, GOOGLE_GENAI_USE_VERTEXAI, +BK_CYMBAL_MODEL, BK_DATA_AGENT_ID) comes from environment variables exported to +~/.bashrc by bk-bootstrap in Lab 1. +""" + +import os + +from google.adk.agents import Agent + +from .ca_tool import ask_cymbal_data_agent + +root_agent = Agent( + name="cymbal_analyst", + model=os.environ.get("BK_CYMBAL_MODEL", "gemini-2.5-flash"), + description=( + "Analytics specialist for Cymbal's governed data warehouse. Answers " + "aggregate business questions (revenue, trends, customer lifetime " + "value, product performance) by querying the gold layer through " + "Cymbal's BigQuery data agent." + ), + instruction=( + "You are Cymbal's analytics specialist.\n" + "For every analytical question, call the ask_cymbal_data_agent tool " + "and base your answer strictly on what it returns -- never invent " + "numbers. Summarize results in clear business language, mention the " + "time range and currency where relevant, and say so explicitly if " + "the data agent could not answer.\n" + "You only handle aggregate analytics on the gold layer. You cannot " + "look up individual live orders -- if asked, say that the concierge " + "handles operational lookups." + ), + tools=[ask_cymbal_data_agent], +) diff --git a/content/agenticdata/src/adk/cymbal_analyst/ca_tool.py b/content/agenticdata/src/adk/cymbal_analyst/ca_tool.py new file mode 100644 index 0000000..6f32b18 --- /dev/null +++ b/content/agenticdata/src/adk/cymbal_analyst/ca_tool.py @@ -0,0 +1,57 @@ +"""Tool that forwards a question to Cymbal's published BigQuery data agent +via the Conversational Analytics API (geminidataanalytics, GA since 2026-06) +and returns the agent's textual answer. + +This is deliberately a small, readable function tool so participants can see +the seams of the integration. ADK also ships a built-in DataAgentToolset +(google.adk.tools.data_agent) that wraps the same API -- swapping to it is a +lab challenge. +""" + +import os + +from google.cloud import geminidataanalytics + + +def ask_cymbal_data_agent(question: str) -> str: + """Ask Cymbal's BigQuery data agent an analytical question about the + governed gold-layer data (revenue, customers, product performance). + + Args: + question: A natural-language business/analytics question. + + Returns: + The data agent's final textual answer (it runs SQL on BigQuery + behind the scenes). + """ + project = os.environ["GOOGLE_CLOUD_PROJECT"] + agent_id = os.environ.get("BK_DATA_AGENT_ID", "cymbal-data-agent") + + client = geminidataanalytics.DataChatServiceClient() + + message = geminidataanalytics.Message() + message.user_message.text = question + + # Stateless chat: reference the published agent's context; no server-side + # conversation resource to manage. + context = geminidataanalytics.DataAgentContext() + context.data_agent = f"projects/{project}/locations/global/dataAgents/{agent_id}" + + request = geminidataanalytics.ChatRequest( + parent=f"projects/{project}/locations/global", + messages=[message], + data_agent_context=context, + ) + + # chat() streams Message objects (thoughts, generated SQL, data, text). + # Collect the text parts of system messages into the final answer. + parts = [] + for reply in client.chat(request=request): + system_message = getattr(reply, "system_message", None) + if system_message is None: + continue + text = getattr(system_message, "text", None) + if text and getattr(text, "parts", None): + parts.extend(text.parts) + + return "\n".join(parts) if parts else "The data agent returned no text answer." diff --git a/content/agenticdata/src/adk/cymbal_concierge/__init__.py b/content/agenticdata/src/adk/cymbal_concierge/__init__.py new file mode 100644 index 0000000..e7186e1 --- /dev/null +++ b/content/agenticdata/src/adk/cymbal_concierge/__init__.py @@ -0,0 +1 @@ +from . import agent # noqa: F401 (ADK discovers root_agent via this import) diff --git a/content/agenticdata/src/adk/cymbal_concierge/agent.py b/content/agenticdata/src/adk/cymbal_concierge/agent.py new file mode 100644 index 0000000..48cd0ea --- /dev/null +++ b/content/agenticdata/src/adk/cymbal_concierge/agent.py @@ -0,0 +1,122 @@ +"""cymbal_concierge -- the root agent participants chat with in `adk web`. + +It spans two planes: + * analytical: delegates aggregate questions over A2A to cymbal_analyst + (which consults the BigQuery data agent on the governed gold layer); + * operational: looks up individual live orders directly in Cloud SQL + Postgres through the IAP tunnel (localhost:5432). + +All configuration (model, database password, hosts) comes from environment +variables exported to ~/.bashrc by bk-bootstrap in Lab 1. +""" + +import os + +import psycopg +from google.adk.agents import Agent +from google.adk.agents.remote_a2a_agent import ( + AGENT_CARD_WELL_KNOWN_PATH, + RemoteA2aAgent, +) + +cymbal_analyst = RemoteA2aAgent( + name="cymbal_analyst", + description=( + "Remote analytics specialist (via the A2A protocol). Handles " + "aggregate business questions: revenue, trends, customer lifetime " + "value, product performance -- anything answered from the governed " + "gold layer in BigQuery." + ), + agent_card=f"http://localhost:8001{AGENT_CARD_WELL_KNOWN_PATH}", +) + + +def check_order_status(order_id: int) -> dict: + """Look up ONE order in Cymbal's LIVE operational Postgres database. + + Use this for questions about a specific order (status, when it was + placed, its total, number of items, payment state). Do NOT use it for + aggregate/analytical questions. + + Args: + order_id: The numeric order id, e.g. 421337. + + Returns: + A dict with the order's live status, or found=False if it does not + exist (it may have been abandoned and deleted). + """ + conn = psycopg.connect( + host=os.environ.get("BK_CYMBAL_DB_HOST", "localhost"), + port=int(os.environ.get("BK_CYMBAL_DB_PORT", "5432")), + dbname="cymbal", + user="postgres", + password=os.environ["BK_DB_PASSWORD"], + connect_timeout=10, + ) + try: + with conn.cursor() as cur: + cur.execute( + """ + SELECT + o.order_id, + o.status, + o.currency, + o.order_ts, + o.updated_at, + COUNT(i.order_item_id) AS items, + COALESCE(SUM(CASE WHEN i.qty > 0 THEN i.qty * i.unit_price END), 0) AS order_total, + MAX(p.status) AS payment_status + FROM cymbal.orders o + LEFT JOIN cymbal.order_items i USING (order_id) + LEFT JOIN cymbal.payments p USING (order_id) + WHERE o.order_id = %s + GROUP BY o.order_id, o.status, o.currency, o.order_ts, o.updated_at + """, + (order_id,), + ) + row = cur.fetchone() + finally: + conn.close() + + if row is None: + return {"found": False, "order_id": order_id, + "note": "No such order in the live database (it may have been abandoned and deleted)."} + return { + "found": True, + "order_id": row[0], + "status": row[1], + "currency": row[2], + "ordered_at": row[3].isoformat(), + "last_updated": row[4].isoformat(), + "items": row[5], + "order_total": float(row[6]), + "payment_status": row[7], + } + + +root_agent = Agent( + name="cymbal_concierge", + model=os.environ.get("BK_CYMBAL_MODEL", "gemini-2.5-flash"), + description="Cymbal's front-desk agent: live order lookups plus delegated analytics.", + instruction=( + "You are the Cymbal concierge.\n" + "Routing rules:\n" + "1. Questions about a SPECIFIC order (status, total, payment of order " + "N) -> call the check_order_status tool. This reads the LIVE " + "operational database.\n" + "2. AGGREGATE or analytical questions (revenue, trends, best " + "customers, product performance, forecasts) -> delegate to the " + "cymbal_analyst sub-agent. It queries the governed gold layer via " + "the BigQuery data agent, so its numbers are as fresh as the CDC " + "pipeline.\n" + "3. If ONE question mixes both planes (a specific order AND an " + "aggregate part), FIRST call check_order_status and answer that part " + "yourself, and only THEN transfer the remaining analytics part to " + "cymbal_analyst.\n" + "Never invent data. If a lookup finds nothing, say so plainly. Keep " + "answers short and business-friendly, and mention which plane " + "(live database vs analytics warehouse) an answer came from." + ), + tools=[check_order_status], + sub_agents=[cymbal_analyst], +) diff --git a/content/agenticdata/src/dataform/AGENTS.md b/content/agenticdata/src/dataform/AGENTS.md new file mode 100644 index 0000000..013d9f5 --- /dev/null +++ b/content/agenticdata/src/dataform/AGENTS.md @@ -0,0 +1,61 @@ +# Cymbal Dataform project — agent context + +This is a Dataform Core 3 project (`workflow_settings.yaml`). It builds a +medallion architecture on BigQuery from a Datastream CDC replica of the +Postgres schema `cymbal`. + +## Source data (bronze, dataset `cymbal_bronze`, declared in definitions/sources.js) + +- `cymbal_customers` (customer_id PK, full_name, email, phone, address, country, created_at, updated_at) +- `cymbal_products` (product_id PK, sku, name, category, price, cost) +- `cymbal_orders` (order_id PK, customer_id, status, currency, order_ts, updated_at) +- `cymbal_order_items` (order_item_id PK, order_id, product_id, qty, unit_price) +- `cymbal_payments` (payment_id PK, order_id, method, amount, status, paid_at) +- `cymbal_reviews` (review_id PK, order_id, rating, review_text, created_at) + +Every bronze table also carries a Datastream `datastream_metadata` column — +exclude it from all downstream tables. + +Known data problems, planted upstream on purpose: duplicate customers (same +email in different letter case), invalid email addresses, empty-string +countries, the status typo `shiped`, mixed-case currency codes, `qty <= 0`, +orphaned order_items (order_id without a matching order), future `order_ts`, +negative payment amounts. + +## Target architecture + +**Silver** (dataset `cymbal_silver`, tag `silver`) — one `stg_*` table per +source table except reviews, fixing the problems above: + +- `stg_customers`: lowercase/trim emails, drop rows whose email is not a + valid address, collapse duplicate emails keeping the most recently updated + row, convert empty-string countries to NULL. +- `stg_products`: cast price/cost to NUMERIC. +- `stg_orders`: lowercase statuses and fix `shiped` -> `shipped`, uppercase + currency codes, drop orders with a future `order_ts`. The complete set of + valid statuses after cleaning is exactly: `pending`, `paid`, `shipped`, + `delivered`, `cancelled`, `returned`. +- `stg_order_items`: drop orphans (inner join to `stg_orders`) and rows with + `qty <= 0`. +- `stg_payments`: drop negative amounts and payments without a matching order. + +**Gold** (dataset `cymbal_gold`, tag `gold`): + +- `fct_daily_revenue`: order_date, currency, distinct orders, units, + gross_revenue = SUM(qty * unit_price), excluding cancelled orders. +- `dim_customer_360`: one row per customer with lifetime_orders, + lifetime_value, avg_order_value, first/last order date. +- `fct_product_performance`: units_sold, gross_revenue and gross_margin = + SUM(qty * (unit_price - cost)) per product, excluding cancelled and + returned orders. + +## Conventions + +- Use `${ref(...)}` for every dependency; never hardcode table names. +- Every silver table gets assertions: `uniqueKey` on its primary key, + `nonNull` on required columns, and `rowConditions` proving the cleaning + worked (allowed status values, currency matches `^[A-Z]{3}$`, `qty > 0`, + `amount >= 0`, `order_ts <= CURRENT_TIMESTAMP()`). +- After changing files, run `dataform compile` and fix errors until the + project compiles cleanly. +- Do not modify `workflow_settings.yaml` or `definitions/sources.js`. diff --git a/content/agenticdata/src/dataform/definitions/sources.js b/content/agenticdata/src/dataform/definitions/sources.js new file mode 100644 index 0000000..6bef932 --- /dev/null +++ b/content/agenticdata/src/dataform/definitions/sources.js @@ -0,0 +1,20 @@ +// Bronze layer: the raw CDC replica maintained by Datastream (merge mode). +// Datastream's "single target dataset" mode names tables _, +// hence cymbal_customers etc. inside the cymbal_bronze dataset. +// Declaring them here lets every model use ${ref(...)} so Dataform tracks +// the dependency graph (and BigQuery lineage sees the whole flow). +const BRONZE_TABLES = [ + "customers", + "products", + "orders", + "order_items", + "payments", + "reviews", +]; + +BRONZE_TABLES.forEach((t) => + declare({ + schema: "cymbal_bronze", + name: `cymbal_${t}`, + }) +); diff --git a/content/agenticdata/src/dataform/workflow_settings.yaml b/content/agenticdata/src/dataform/workflow_settings.yaml new file mode 100644 index 0000000..0ad6786 --- /dev/null +++ b/content/agenticdata/src/dataform/workflow_settings.yaml @@ -0,0 +1,9 @@ +# Cymbal medallion pipeline -- Dataform Core project settings. +# PROJECT_ID_PLACEHOLDER is replaced in the lab: +# sed -i "s/PROJECT_ID_PLACEHOLDER/$PROJECT_ID/" workflow_settings.yaml +# Keep dataformCoreVersion aligned with the pinned @dataform/cli major version. +defaultProject: PROJECT_ID_PLACEHOLDER +defaultLocation: US +defaultDataset: cymbal_silver +defaultAssertionDataset: cymbal_assertions +dataformCoreVersion: 3.0.0 diff --git a/content/agenticdata/src/dataform_reference/definitions/assert_bronze_payments_non_negative.sqlx b/content/agenticdata/src/dataform_reference/definitions/assert_bronze_payments_non_negative.sqlx new file mode 100644 index 0000000..dbf89e1 --- /dev/null +++ b/content/agenticdata/src/dataform_reference/definitions/assert_bronze_payments_non_negative.sqlx @@ -0,0 +1,12 @@ +config { + type: "assertion", + tags: ["challenge"], + description: "CHALLENGE reference: detects the planted negative payment amounts in BRONZE instead of silently filtering them in silver. This assertion is EXPECTED TO FAIL on the seed data -- that is the point: silver quarantines, assertions alert. Deliberately tagged 'challenge' (not 'silver') so the regular silver run stays green; run it with `dataform run --tags challenge`." +} + +SELECT + payment_id, + order_id, + amount +FROM ${ref("cymbal_payments")} +WHERE amount < 0 diff --git a/content/agenticdata/src/dataform_reference/definitions/dim_customer_360.sqlx b/content/agenticdata/src/dataform_reference/definitions/dim_customer_360.sqlx new file mode 100644 index 0000000..0ab8c1c --- /dev/null +++ b/content/agenticdata/src/dataform_reference/definitions/dim_customer_360.sqlx @@ -0,0 +1,42 @@ +config { + type: "table", + schema: "cymbal_gold", + tags: ["gold"], + description: "One row per (deduplicated) customer with lifetime metrics. 'Lifetime value' here means gross revenue of non-cancelled orders.", + assertions: { + uniqueKey: ["customer_id"] + }, + columns: { + lifetime_orders: "Count of non-cancelled orders", + lifetime_value: "Gross revenue across all non-cancelled orders", + avg_order_value: "lifetime_value / lifetime_orders" + } +} + +WITH order_totals AS ( + SELECT + o.order_id, + o.customer_id, + o.order_ts, + SUM(i.qty * i.unit_price) AS order_total + FROM ${ref("stg_orders")} AS o + INNER JOIN ${ref("stg_order_items")} AS i + USING (order_id) + WHERE o.status != 'cancelled' + GROUP BY o.order_id, o.customer_id, o.order_ts +) + +SELECT + c.customer_id, + c.full_name, + c.email, + c.country, + COUNT(t.order_id) AS lifetime_orders, + ROUND(COALESCE(SUM(t.order_total), 0), 2) AS lifetime_value, + ROUND(COALESCE(SUM(t.order_total), 0) / NULLIF(COUNT(t.order_id), 0), 2) AS avg_order_value, + MIN(DATE(t.order_ts)) AS first_order_date, + MAX(DATE(t.order_ts)) AS last_order_date +FROM ${ref("stg_customers")} AS c +LEFT JOIN order_totals AS t + USING (customer_id) +GROUP BY c.customer_id, c.full_name, c.email, c.country diff --git a/content/agenticdata/src/dataform_reference/definitions/fct_daily_revenue.sqlx b/content/agenticdata/src/dataform_reference/definitions/fct_daily_revenue.sqlx new file mode 100644 index 0000000..2acc670 --- /dev/null +++ b/content/agenticdata/src/dataform_reference/definitions/fct_daily_revenue.sqlx @@ -0,0 +1,25 @@ +config { + type: "table", + schema: "cymbal_gold", + tags: ["gold"], + description: "Daily gross revenue and volume per currency, excluding cancelled orders. Primary mart for the Cymbal data agent.", + columns: { + order_date: "Calendar date of the order (UTC)", + currency: "ISO 4217 currency code, normalized in silver", + orders: "Distinct orders placed that day", + units: "Total units sold", + gross_revenue: "SUM(qty * unit_price) before refunds and costs" + } +} + +SELECT + DATE(o.order_ts) AS order_date, + o.currency, + COUNT(DISTINCT o.order_id) AS orders, + SUM(i.qty) AS units, + ROUND(SUM(i.qty * i.unit_price), 2) AS gross_revenue +FROM ${ref("stg_orders")} AS o +INNER JOIN ${ref("stg_order_items")} AS i + USING (order_id) +WHERE o.status != 'cancelled' +GROUP BY order_date, currency diff --git a/content/agenticdata/src/dataform_reference/definitions/fct_product_performance.sqlx b/content/agenticdata/src/dataform_reference/definitions/fct_product_performance.sqlx new file mode 100644 index 0000000..3757708 --- /dev/null +++ b/content/agenticdata/src/dataform_reference/definitions/fct_product_performance.sqlx @@ -0,0 +1,25 @@ +config { + type: "table", + schema: "cymbal_gold", + tags: ["gold"], + description: "Units, revenue and gross margin per product, excluding cancelled and returned orders.", + assertions: { + uniqueKey: ["product_id"] + } +} + +SELECT + p.product_id, + p.sku, + p.name, + p.category, + SUM(i.qty) AS units_sold, + ROUND(SUM(i.qty * i.unit_price), 2) AS gross_revenue, + ROUND(SUM(i.qty * (i.unit_price - p.cost)), 2) AS gross_margin +FROM ${ref("stg_order_items")} AS i +INNER JOIN ${ref("stg_products")} AS p + USING (product_id) +INNER JOIN ${ref("stg_orders")} AS o + USING (order_id) +WHERE o.status NOT IN ('cancelled', 'returned') +GROUP BY p.product_id, p.sku, p.name, p.category diff --git a/content/agenticdata/src/dataform_reference/definitions/stg_customers.sqlx b/content/agenticdata/src/dataform_reference/definitions/stg_customers.sqlx new file mode 100644 index 0000000..35fd269 --- /dev/null +++ b/content/agenticdata/src/dataform_reference/definitions/stg_customers.sqlx @@ -0,0 +1,40 @@ +config { + type: "table", + schema: "cymbal_silver", + tags: ["silver"], + description: "Cleaned, deduplicated customers. Invalid emails are dropped (quarantine-by-filter), emails are normalized to lowercase, and duplicates (same email in different letter case) are collapsed to the most recently updated record.", + assertions: { + uniqueKey: ["customer_id"], + nonNull: ["customer_id", "full_name", "email"] + } +} + +WITH cleaned AS ( + SELECT + customer_id, + full_name, + LOWER(TRIM(email)) AS email, + phone, + address, + NULLIF(TRIM(country), '') AS country, + created_at, + updated_at + FROM ${ref("cymbal_customers")} + WHERE REGEXP_CONTAINS(TRIM(email), r'^[^@\s]+@[^@\s]+\.[^@\s]+$') +), + +deduped AS ( + SELECT * EXCEPT (rn) + FROM ( + SELECT + *, + ROW_NUMBER() OVER ( + PARTITION BY email + ORDER BY updated_at DESC, customer_id + ) AS rn + FROM cleaned + ) + WHERE rn = 1 +) + +SELECT * FROM deduped diff --git a/content/agenticdata/src/dataform_reference/definitions/stg_order_items.sqlx b/content/agenticdata/src/dataform_reference/definitions/stg_order_items.sqlx new file mode 100644 index 0000000..3ead2f4 --- /dev/null +++ b/content/agenticdata/src/dataform_reference/definitions/stg_order_items.sqlx @@ -0,0 +1,25 @@ +config { + type: "table", + schema: "cymbal_silver", + tags: ["silver"], + description: "Order items with orphans dropped (inner join to stg_orders removes the planted referential drift) and non-positive quantities filtered out.", + assertions: { + uniqueKey: ["order_item_id"], + nonNull: ["order_item_id", "order_id", "product_id"], + rowConditions: [ + "qty > 0", + "unit_price > 0" + ] + } +} + +SELECT + i.order_item_id, + i.order_id, + i.product_id, + i.qty, + CAST(i.unit_price AS NUMERIC) AS unit_price +FROM ${ref("cymbal_order_items")} AS i +INNER JOIN ${ref("stg_orders")} AS o + USING (order_id) +WHERE i.qty > 0 diff --git a/content/agenticdata/src/dataform_reference/definitions/stg_orders.sqlx b/content/agenticdata/src/dataform_reference/definitions/stg_orders.sqlx new file mode 100644 index 0000000..5b2f297 --- /dev/null +++ b/content/agenticdata/src/dataform_reference/definitions/stg_orders.sqlx @@ -0,0 +1,28 @@ +config { + type: "table", + schema: "cymbal_silver", + tags: ["silver"], + description: "Orders with normalized statuses (fixes the planted 'shiped' typo), uppercased ISO currency codes, and future-dated rows filtered out.", + assertions: { + uniqueKey: ["order_id"], + nonNull: ["order_id", "customer_id", "status", "currency"], + rowConditions: [ + "status IN ('pending', 'paid', 'shipped', 'delivered', 'cancelled', 'returned')", + "REGEXP_CONTAINS(currency, r'^[A-Z]{3}$')", + "order_ts <= CURRENT_TIMESTAMP()" + ] + } +} + +SELECT + order_id, + customer_id, + CASE LOWER(TRIM(status)) + WHEN 'shiped' THEN 'shipped' + ELSE LOWER(TRIM(status)) + END AS status, + UPPER(TRIM(currency)) AS currency, + order_ts, + updated_at +FROM ${ref("cymbal_orders")} +WHERE order_ts <= CURRENT_TIMESTAMP() diff --git a/content/agenticdata/src/dataform_reference/definitions/stg_payments.sqlx b/content/agenticdata/src/dataform_reference/definitions/stg_payments.sqlx new file mode 100644 index 0000000..ff587d4 --- /dev/null +++ b/content/agenticdata/src/dataform_reference/definitions/stg_payments.sqlx @@ -0,0 +1,25 @@ +config { + type: "table", + schema: "cymbal_silver", + tags: ["silver"], + description: "Payments joined to valid orders; planted negative amounts are filtered out (see the challenge: an assertion on the BRONZE table detects them instead of silently dropping them).", + assertions: { + uniqueKey: ["payment_id"], + nonNull: ["payment_id", "order_id", "method", "status"], + rowConditions: [ + "amount >= 0" + ] + } +} + +SELECT + p.payment_id, + p.order_id, + p.method, + CAST(p.amount AS NUMERIC) AS amount, + p.status, + p.paid_at +FROM ${ref("cymbal_payments")} AS p +INNER JOIN ${ref("stg_orders")} AS o + USING (order_id) +WHERE p.amount >= 0 diff --git a/content/agenticdata/src/dataform_reference/definitions/stg_products.sqlx b/content/agenticdata/src/dataform_reference/definitions/stg_products.sqlx new file mode 100644 index 0000000..5e70c11 --- /dev/null +++ b/content/agenticdata/src/dataform_reference/definitions/stg_products.sqlx @@ -0,0 +1,23 @@ +config { + type: "table", + schema: "cymbal_silver", + tags: ["silver"], + description: "Products with numeric types normalized and a sanity margin check.", + assertions: { + uniqueKey: ["product_id"], + nonNull: ["product_id", "sku", "name", "category"], + rowConditions: [ + "price > 0", + "cost > 0" + ] + } +} + +SELECT + product_id, + sku, + name, + category, + CAST(price AS NUMERIC) AS price, + CAST(cost AS NUMERIC) AS cost +FROM ${ref("cymbal_products")} diff --git a/content/agenticdata/src/datagen/generate.py b/content/agenticdata/src/datagen/generate.py new file mode 100755 index 0000000..979b51f --- /dev/null +++ b/content/agenticdata/src/datagen/generate.py @@ -0,0 +1,252 @@ +#!/usr/bin/env python3 +"""Deterministic synthetic OLTP data generator for the Cymbal orders platform. + +Writes header-less CSV files (one per table, column order matching schema.sql) +suitable for `gcloud sql import csv` (server-side COPY ... CSV). With the fixed +seed and pinned dependency versions, every participant generates *identical* +data, so lab instructions, data-quality thresholds and verified queries can +reference exact values. + +The data contains PLANTED FLAWS on purpose -- they are the raw material for +the silver layer and the Knowledge Catalog data-quality lab: + + * duplicate customers (same email, different letter case, new customer_id) + * invalid email addresses + * NULL countries + * order status typo: 'shiped' + * mixed-case currency codes ('eur' next to 'EUR') + * zero/negative order_items.qty + * orphaned order_items (order_id that does not exist) + * orders with a future order_ts + * negative payment amounts + +Usage: + python3 generate.py --out ~/seed_data [--scale 1.0] +""" + +import argparse +import csv +import os +import random +from datetime import datetime, timedelta, timezone + +from faker import Faker + +SEED = 42 +WINDOW_START = datetime(2025, 1, 1, tzinfo=timezone.utc) +WINDOW_DAYS = 540 # Jan 2025 .. mid 2026 + +N_CUSTOMERS = 50_000 +N_PRODUCTS = 5_000 +N_ORDERS = 500_000 +N_REVIEWS = 50_000 + +CATEGORIES = [ + "Home & Kitchen", "Electronics", "Outdoor", "Office", "Toys & Games", + "Beauty", "Sports", "Automotive", "Pet Supplies", "Grocery", + "Apparel", "Garden", +] +COUNTRIES = [ + "Germany", "France", "Netherlands", "Spain", "Italy", "Poland", + "Belgium", "Austria", "Switzerland", "Sweden", "Portugal", "Ireland", +] +EMAIL_DOMAINS = ["example.com", "example.org", "mail.example.net", "inbox.example.dev"] +CURRENCIES = ["EUR", "USD", "GBP", "CHF", "PLN"] +CURRENCY_WEIGHTS = [50, 25, 10, 10, 5] +STATUSES = ["pending", "paid", "shipped", "delivered", "cancelled", "returned"] +STATUS_WEIGHTS = [5, 10, 15, 55, 10, 5] +PAY_METHODS = ["card", "paypal", "bank_transfer", "gift_card"] +PRODUCT_ADJECTIVES = [ + "Ultra", "Compact", "Deluxe", "Eco", "Smart", "Classic", "Portable", + "Premium", "Foldable", "Wireless", "Heavy-Duty", "Mini", +] +PRODUCT_NOUNS = [ + "Blender", "Lamp", "Backpack", "Speaker", "Notebook", "Bottle", "Chair", + "Charger", "Kettle", "Tent", "Keyboard", "Mat", "Drill", "Mug", "Router", +] +REVIEW_TEMPLATES = [ + "Absolutely love this {noun}. {extra}", + "The {noun} arrived quickly and works as described. {extra}", + "Not what I expected. The {noun} feels cheap. {extra}", + "Great value for money, would buy this {noun} again. {extra}", + "The {noun} stopped working after a week. Support was {support}.", + "Five stars. Best {noun} I have owned so far. {extra}", + "Mediocre {noun}. Delivery was {support}, though.", +] +REVIEW_EXTRAS = [ + "Highly recommended.", "Shipping took longer than promised.", + "Packaging was damaged but the product is fine.", "My whole family uses it.", + "Will order again.", "Returned it the next day.", "Setup took two minutes.", +] +SUPPORT_WORDS = ["helpful", "slow", "excellent", "unresponsive", "friendly"] + +# Planted-flaw rates +P_DUP_CUSTOMER = 0.010 +P_INVALID_EMAIL = 0.005 +P_NULL_COUNTRY = 0.015 +P_STATUS_TYPO = 0.040 # applied to 'shipped' orders only +P_CURRENCY_LOWER = 0.030 +P_BAD_QTY = 0.003 +P_ORPHAN_ITEM = 0.002 +P_FUTURE_TS = 0.001 +P_NEGATIVE_PAYMENT = 0.002 + +ORPHAN_ORDER_BASE = 90_000_000 + + +def fmt(dt): + return dt.strftime("%Y-%m-%d %H:%M:%S+00") + + +def rand_ts(rnd): + return WINDOW_START + timedelta(seconds=rnd.randint(0, WINDOW_DAYS * 86400 - 1)) + + +def writer(out_dir, name): + f = open(os.path.join(out_dir, name), "w", newline="") + return f, csv.writer(f) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--out", default="./seed_data", help="output directory for CSV files") + parser.add_argument("--scale", type=float, default=1.0, + help="scale factor for row counts (use e.g. 0.1 for a quick re-run)") + args = parser.parse_args() + + os.makedirs(args.out, exist_ok=True) + rnd = random.Random(SEED) + Faker.seed(SEED) + fake = Faker() + + n_customers = int(N_CUSTOMERS * args.scale) + n_products = int(N_PRODUCTS * args.scale) + n_orders = int(N_ORDERS * args.scale) + n_reviews = min(int(N_REVIEWS * args.scale), n_orders) + + # Small Faker-built pools; rows are assembled from pools (fast + deterministic). + print("Building value pools ...") + first_names = [fake.first_name() for _ in range(600)] + last_names = [fake.last_name() for _ in range(600)] + streets = [fake.street_name() for _ in range(300)] + cities = [fake.city() for _ in range(200)] + + # ------------------------------------------------------------- customers + print(f"Generating {n_customers} customers (+ ~{P_DUP_CUSTOMER:.0%} duplicates) ...") + f, w = writer(args.out, "customers.csv") + duplicates = [] + for cid in range(1, n_customers + 1): + fn, ln = rnd.choice(first_names), rnd.choice(last_names) + full_name = f"{fn} {ln}" + email = f"{fn}.{ln}{cid}@{rnd.choice(EMAIL_DOMAINS)}".lower() + if rnd.random() < P_INVALID_EMAIL: + email = email.replace("@", " at ") + phone = f"+{rnd.randint(1, 49)} {rnd.randint(100, 999)} {rnd.randint(1_000_000, 9_999_999)}" + address = f"{rnd.randint(1, 999)} {rnd.choice(streets)}, {rnd.choice(cities)}" + country = "" if rnd.random() < P_NULL_COUNTRY else rnd.choice(COUNTRIES) + created = rand_ts(rnd) + updated = created + timedelta(seconds=rnd.randint(0, 90 * 86400)) + w.writerow([cid, full_name, email, phone, address, country, fmt(created), fmt(updated)]) + if rnd.random() < P_DUP_CUSTOMER: + duplicates.append((full_name, email, phone, country, created)) + # Duplicate customers: same person, email in a different letter case, new id. + next_id = n_customers + 1 + for full_name, email, phone, country, created in duplicates: + dup_email = email.upper() if rnd.random() < 0.5 else email.title() + address = f"{rnd.randint(1, 999)} {rnd.choice(streets)}, {rnd.choice(cities)}" + created2 = created + timedelta(seconds=rnd.randint(3600, 200 * 86400)) + w.writerow([next_id, full_name, dup_email, phone, address, country, + fmt(created2), fmt(created2)]) + next_id += 1 + total_customers = next_id - 1 + f.close() + + # -------------------------------------------------------------- products + print(f"Generating {n_products} products ...") + f, w = writer(args.out, "products.csv") + prices = [0.0] * (n_products + 1) + for pid in range(1, n_products + 1): + name = f"{rnd.choice(PRODUCT_ADJECTIVES)} {rnd.choice(PRODUCT_NOUNS)}" + price = round(min(999.0, max(3.0, rnd.lognormvariate(3.2, 0.7))), 2) + cost = round(price * rnd.uniform(0.4, 0.8), 2) + prices[pid] = price + w.writerow([pid, f"CYM-{pid:05d}", name, rnd.choice(CATEGORIES), price, cost]) + f.close() + + # ------------------------------------- orders + order_items + payments + print(f"Generating {n_orders} orders (+ items, payments) ...") + fo, wo = writer(args.out, "orders.csv") + fi, wi = writer(args.out, "order_items.csv") + fp, wp = writer(args.out, "payments.csv") + item_id = 0 + orphan_count = 0 + for oid in range(1, n_orders + 1): + customer_id = rnd.randint(1, total_customers) + order_ts = rand_ts(rnd) + if rnd.random() < P_FUTURE_TS: + order_ts = datetime(2027, 1, 1, tzinfo=timezone.utc) + timedelta( + seconds=rnd.randint(0, 180 * 86400)) + status = rnd.choices(STATUSES, weights=STATUS_WEIGHTS, k=1)[0] + if status == "shipped" and rnd.random() < P_STATUS_TYPO: + status = "shiped" + currency = rnd.choices(CURRENCIES, weights=CURRENCY_WEIGHTS, k=1)[0] + if rnd.random() < P_CURRENCY_LOWER: + currency = currency.lower() + updated = order_ts + timedelta(seconds=rnd.randint(0, 14 * 86400)) + wo.writerow([oid, customer_id, status, currency, fmt(order_ts), fmt(updated)]) + + n_items = rnd.choices([1, 2, 3, 4, 5], weights=[30, 30, 20, 12, 8], k=1)[0] + order_total = 0.0 + for _ in range(n_items): + item_id += 1 + product_id = rnd.randint(1, n_products) + qty = rnd.randint(1, 5) + if rnd.random() < P_BAD_QTY: + qty = rnd.choice([0, -1, -2]) + unit_price = round(prices[product_id] * rnd.uniform(0.9, 1.1), 2) + parent = oid + if rnd.random() < P_ORPHAN_ITEM: + orphan_count += 1 + parent = ORPHAN_ORDER_BASE + orphan_count + wi.writerow([item_id, parent, product_id, qty, unit_price]) + if parent == oid and qty > 0: + order_total += qty * unit_price + + amount = round(order_total, 2) + if rnd.random() < P_NEGATIVE_PAYMENT: + amount = -amount + pay_status = "captured" if status in ("paid", "shipped", "shiped", "delivered", "returned") else ( + "void" if status == "cancelled" else "pending") + paid_at = fmt(order_ts + timedelta(minutes=rnd.randint(1, 240))) if pay_status == "captured" else "" + wp.writerow([oid, oid, rnd.choice(PAY_METHODS), amount, pay_status, paid_at]) + + if oid % 100_000 == 0: + print(f" ... {oid} orders") + fo.close(); fi.close(); fp.close() + + # --------------------------------------------------------------- reviews + print(f"Generating {n_reviews} reviews ...") + f, w = writer(args.out, "reviews.csv") + reviewed = rnd.sample(range(1, n_orders + 1), n_reviews) + for rid, order_id in enumerate(sorted(reviewed), start=1): + rating = rnd.choices([1, 2, 3, 4, 5], weights=[6, 8, 16, 35, 35], k=1)[0] + text = rnd.choice(REVIEW_TEMPLATES).format( + noun=rnd.choice(PRODUCT_NOUNS).lower(), + extra=rnd.choice(REVIEW_EXTRAS), + support=rnd.choice(SUPPORT_WORDS), + ) + created = rand_ts(rnd) + w.writerow([rid, order_id, rating, text, fmt(created)]) + f.close() + + print(f"\nDone. CSV files written to {os.path.abspath(args.out)}") + print(f" customers: {total_customers} (incl. {total_customers - n_customers} planted duplicates)") + print(f" products: {n_products}") + print(f" orders: {n_orders}") + print(f" order_items: {item_id} (incl. {orphan_count} planted orphans)") + print(f" payments: {n_orders}") + print(f" reviews: {n_reviews}") + + +if __name__ == "__main__": + main() diff --git a/content/agenticdata/src/datagen/schema.sql b/content/agenticdata/src/datagen/schema.sql new file mode 100644 index 0000000..8769ead --- /dev/null +++ b/content/agenticdata/src/datagen/schema.sql @@ -0,0 +1,64 @@ +-- Cymbal OLTP schema (Agentic Data Bootkon). +-- +-- Every table has a single-column integer primary key: Datastream's BigQuery +-- merge mode requires primary keys (and forbids FLOAT/REAL keys). +-- +-- There are deliberately NO foreign-key constraints: the seed data contains +-- planted referential drift (orphaned order_items) that is part of the +-- data-quality story told in the Dataform and Knowledge Catalog labs. + +CREATE SCHEMA IF NOT EXISTS cymbal; + +CREATE TABLE cymbal.customers ( + customer_id BIGINT PRIMARY KEY, + full_name TEXT NOT NULL, + email TEXT, + phone TEXT, + address TEXT, + country TEXT, + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL +); + +CREATE TABLE cymbal.products ( + product_id BIGINT PRIMARY KEY, + sku TEXT NOT NULL, + name TEXT NOT NULL, + category TEXT NOT NULL, + price NUMERIC(10, 2) NOT NULL, + cost NUMERIC(10, 2) NOT NULL +); + +CREATE TABLE cymbal.orders ( + order_id BIGINT PRIMARY KEY, + customer_id BIGINT NOT NULL, + status TEXT NOT NULL, + currency TEXT NOT NULL, + order_ts TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL +); + +CREATE TABLE cymbal.order_items ( + order_item_id BIGINT PRIMARY KEY, + order_id BIGINT NOT NULL, + product_id BIGINT NOT NULL, + qty INTEGER NOT NULL, + unit_price NUMERIC(10, 2) NOT NULL +); + +CREATE TABLE cymbal.payments ( + payment_id BIGINT PRIMARY KEY, + order_id BIGINT NOT NULL, + method TEXT NOT NULL, + amount NUMERIC(12, 2) NOT NULL, + status TEXT NOT NULL, + paid_at TIMESTAMPTZ +); + +CREATE TABLE cymbal.reviews ( + review_id BIGINT PRIMARY KEY, + order_id BIGINT NOT NULL, + rating INTEGER NOT NULL, + review_text TEXT, + created_at TIMESTAMPTZ NOT NULL +); diff --git a/content/agenticdata/src/datagen/simulate.py b/content/agenticdata/src/datagen/simulate.py new file mode 100755 index 0000000..b240817 --- /dev/null +++ b/content/agenticdata/src/datagen/simulate.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +"""Live-activity simulator for the Cymbal orders platform. + +Continuously INSERTs new orders (with items and a payment), UPDATEs order +statuses and customer emails, and occasionally DELETEs an abandoned pending +order -- so Datastream always has change events to replicate. Run it in its +own terminal and leave it running for the rest of the event. + +It connects through the IAP tunnel, so the tunnel terminal must stay open: + + python3 content/agenticdata/src/datagen/simulate.py + +Environment: BK_DB_PASSWORD (required), BK_CYMBAL_DB_HOST (default localhost), +BK_CYMBAL_DB_PORT (default 5432). +""" + +import os +import random +import sys +import time +from datetime import datetime, timezone + +import psycopg + +CURRENCIES = ["EUR", "USD", "GBP", "CHF", "PLN"] +PAY_METHODS = ["card", "paypal", "bank_transfer", "gift_card"] +STATUS_NEXT = {"pending": "paid", "paid": "shipped", "shipped": "delivered"} +EMAIL_DOMAINS = ["example.com", "example.org", "mail.example.net"] + + +def now(): + return datetime.now(timezone.utc) + + +def log(action, detail): + print(f"[{now().strftime('%H:%M:%S')}] {action:<14} {detail}", flush=True) + + +def main(): + password = os.environ.get("BK_DB_PASSWORD") + if not password: + sys.exit("BK_DB_PASSWORD is not set. Run: source ~/.bashrc (set by bk-bootstrap)") + host = os.environ.get("BK_CYMBAL_DB_HOST", "localhost") + port = int(os.environ.get("BK_CYMBAL_DB_PORT", "5432")) + + conn = psycopg.connect(host=host, port=port, dbname="cymbal", user="postgres", + password=password, connect_timeout=10, autocommit=True) + rnd = random.Random() # intentionally unseeded: every run produces new events + + with conn.cursor() as cur: + cur.execute("SELECT COALESCE(MAX(order_id), 0), " + "(SELECT COALESCE(MAX(order_item_id), 0) FROM cymbal.order_items), " + "(SELECT COALESCE(MAX(payment_id), 0) FROM cymbal.payments), " + "(SELECT COALESCE(MAX(customer_id), 0) FROM cymbal.customers) " + "FROM cymbal.orders") + next_order, next_item, next_payment, max_customer = cur.fetchone() + print(f"Connected to {host}:{port}/cymbal -- simulating live activity. Ctrl+C to stop.") + + inserted = updated = deleted = 0 + try: + while True: + r = rnd.random() + with conn.cursor() as cur: + if r < 0.60: # new order + next_order += 1 + next_payment += 1 + customer_id = rnd.randint(1, max_customer) + currency = rnd.choice(CURRENCIES) + cur.execute( + "INSERT INTO cymbal.orders VALUES (%s, %s, 'pending', %s, %s, %s)", + (next_order, customer_id, currency, now(), now())) + total = 0.0 + for _ in range(rnd.randint(1, 3)): + next_item += 1 + product_id = rnd.randint(1, 5000) + qty = rnd.randint(1, 4) + cur.execute("SELECT price FROM cymbal.products WHERE product_id = %s", + (product_id,)) + row = cur.fetchone() + unit_price = round(float(row[0]) * rnd.uniform(0.9, 1.1), 2) if row else 19.99 + cur.execute("INSERT INTO cymbal.order_items VALUES (%s, %s, %s, %s, %s)", + (next_item, next_order, product_id, qty, unit_price)) + total += qty * unit_price + cur.execute("INSERT INTO cymbal.payments VALUES (%s, %s, %s, %s, 'pending', NULL)", + (next_payment, next_order, rnd.choice(PAY_METHODS), round(total, 2))) + inserted += 1 + log("INSERT order", f"order_id={next_order} total={total:.2f} {currency}") + + elif r < 0.85: # progress an order status + cur.execute( + "SELECT order_id, status FROM cymbal.orders " + "WHERE status IN ('pending', 'paid', 'shipped') " + "ORDER BY order_id DESC LIMIT 50") + rows = cur.fetchall() + if rows: + order_id, status = rnd.choice(rows) + new_status = STATUS_NEXT[status] + cur.execute( + "UPDATE cymbal.orders SET status = %s, updated_at = %s " + "WHERE order_id = %s", (new_status, now(), order_id)) + if new_status == "paid": + cur.execute( + "UPDATE cymbal.payments SET status = 'captured', paid_at = %s " + "WHERE order_id = %s", (now(), order_id)) + updated += 1 + log("UPDATE status", f"order_id={order_id} {status} -> {new_status}") + + elif r < 0.95: # customer changes their email + customer_id = rnd.randint(1, max_customer) + new_email = f"user{customer_id}.{rnd.randint(100, 999)}@{rnd.choice(EMAIL_DOMAINS)}" + cur.execute( + "UPDATE cymbal.customers SET email = %s, updated_at = %s " + "WHERE customer_id = %s", (new_email, now(), customer_id)) + updated += 1 + log("UPDATE email", f"customer_id={customer_id} -> {new_email}") + + else: # abandon (delete) a recent pending order + cur.execute( + "SELECT order_id FROM cymbal.orders WHERE status = 'pending' " + "ORDER BY order_id DESC LIMIT 20") + rows = cur.fetchall() + if rows: + order_id = rnd.choice(rows)[0] + cur.execute("DELETE FROM cymbal.order_items WHERE order_id = %s", (order_id,)) + cur.execute("DELETE FROM cymbal.payments WHERE order_id = %s", (order_id,)) + cur.execute("DELETE FROM cymbal.orders WHERE order_id = %s", (order_id,)) + deleted += 1 + log("DELETE order", f"order_id={order_id} (abandoned)") + + time.sleep(rnd.uniform(1.0, 3.0)) + except KeyboardInterrupt: + print(f"\nStopped. {inserted} orders inserted, {updated} updates, {deleted} deletes.") + + +if __name__ == "__main__": + main() diff --git a/content/agenticdata/src/datastream/destination_config.json b/content/agenticdata/src/datastream/destination_config.json new file mode 100644 index 0000000..d0a2c51 --- /dev/null +++ b/content/agenticdata/src/datastream/destination_config.json @@ -0,0 +1,6 @@ +{ + "singleTargetDataset": { + "datasetId": "PROJECT_ID_PLACEHOLDER:cymbal_bronze" + }, + "dataFreshness": "0s" +} diff --git a/content/agenticdata/src/datastream/replication_setup.sql b/content/agenticdata/src/datastream/replication_setup.sql new file mode 100644 index 0000000..81025e4 --- /dev/null +++ b/content/agenticdata/src/datastream/replication_setup.sql @@ -0,0 +1,29 @@ +-- Datastream logical-replication setup for the Cymbal database. +-- Run through the IAP tunnel with the password passed as a psql variable: +-- +-- PGPASSWORD="$BK_DB_PASSWORD" psql -h localhost -p 5432 -U postgres -d cymbal \ +-- -v ds_password="$BK_DS_PASSWORD" \ +-- -f content/agenticdata/src/datastream/replication_setup.sql +-- +-- Notes: +-- * The replication user follows the documented Cloud SQL pattern +-- (WITH REPLICATION IN ROLE cloudsqlsuperuser). +-- * FOR ALL TABLES avoids the classic "table missing from publication" +-- stream-validation failure. +-- * Names are case-sensitive when referenced at stream creation +-- (cymbal_pub / cymbal_slot -- keep lowercase everywhere). + +-- On Cloud SQL, the default postgres user is cloudsqlsuperuser but does NOT +-- carry the REPLICATION attribute -- without this line, the slot creation +-- below fails with "must be superuser or replication role" (verified live). +ALTER USER postgres WITH REPLICATION; + +CREATE USER datastream_user WITH REPLICATION IN ROLE cloudsqlsuperuser LOGIN PASSWORD :'ds_password'; + +GRANT USAGE ON SCHEMA cymbal TO datastream_user; +GRANT SELECT ON ALL TABLES IN SCHEMA cymbal TO datastream_user; +ALTER DEFAULT PRIVILEGES IN SCHEMA cymbal GRANT SELECT ON TABLES TO datastream_user; + +CREATE PUBLICATION cymbal_pub FOR ALL TABLES; + +SELECT PG_CREATE_LOGICAL_REPLICATION_SLOT('cymbal_slot', 'pgoutput'); diff --git a/content/agenticdata/src/datastream/source_config.json b/content/agenticdata/src/datastream/source_config.json new file mode 100644 index 0000000..05075c3 --- /dev/null +++ b/content/agenticdata/src/datastream/source_config.json @@ -0,0 +1,11 @@ +{ + "publication": "cymbal_pub", + "replicationSlot": "cymbal_slot", + "includeObjects": { + "postgresqlSchemas": [ + { + "schema": "cymbal" + } + ] + } +} diff --git a/content/agenticdata/src/jumpvm-startup.sh b/content/agenticdata/src/jumpvm-startup.sh new file mode 100644 index 0000000..3dbb8de --- /dev/null +++ b/content/agenticdata/src/jumpvm-startup.sh @@ -0,0 +1,8 @@ +#!/bin/bash +# Startup script for the cymbal-jump VM: forwards port 5432 to the Cymbal +# database at the PSC endpoint (10.10.0.5) via iptables DNAT. Linux startup +# scripts run on every boot, so the rules survive VM restarts. +sysctl -w net.ipv4.ip_forward=1 +iptables -t nat -F +iptables -t nat -A PREROUTING -p tcp --dport 5432 -j DNAT --to-destination 10.10.0.5 +iptables -t nat -A POSTROUTING -p tcp --dport 5432 -j MASQUERADE diff --git a/content/agenticdata/src/prompts.md b/content/agenticdata/src/prompts.md new file mode 100644 index 0000000..847b1f3 --- /dev/null +++ b/content/agenticdata/src/prompts.md @@ -0,0 +1,83 @@ +# Canonical agy prompts + +The prompts quoted in the labs, collected in one place so table captains can +copy them quickly. agy output varies by design — every artifact has a +reference solution under `content/agenticdata/src/` that participants can +fall back to. + +## Lab 1 — understand the bootstrap + +``` +What does content/agenticdata/bk-bootstrap do? Summarize the IAM roles it +grants and explain why a separate data-quality service account is created. +``` + +## Lab 2 — Datastream configuration files + +``` +/goal Rewrite content/agenticdata/src/datastream/source_config.json and +content/agenticdata/src/datastream/destination_config.json for +`gcloud datastream streams create`. +The PostgreSQL source uses publication "cymbal_pub", replication slot +"cymbal_slot", and should include all tables of the "cymbal" schema. +The BigQuery destination writes every table into the single dataset +":cymbal_bronze" with a data freshness of 0 seconds. +Use the exact JSON field names of the Datastream v1 API +(publication, replicationSlot, includeObjects/postgresqlSchemas, +singleTargetDataset/datasetId, dataFreshness). +``` + +Fallback: `git restore content/agenticdata/src/datastream/` then replace +`PROJECT_ID_PLACEHOLDER` in `destination_config.json` with sed. + +## Lab 3 — the medallion pipeline (the centerpiece) + +Run inside `~/bootkon/content/agenticdata/src/dataform`. The full +specification lives in that folder's `AGENTS.md` — the prompt only hands +over the brief: + +``` +/goal Read AGENTS.md and build the complete silver and gold layers it +specifies. Run `dataform compile` yourself and fix any errors until the +project compiles cleanly. +``` + +Fallback: `cp content/agenticdata/src/dataform_reference/definitions/*.sqlx content/agenticdata/src/dataform/definitions/` + +## Lab 3 challenge — incremental gold + +``` +/goal Convert fct_daily_revenue into an incremental Dataform table that only +processes orders newer than the latest order_date already in the table. +Explain the trade-off versus a full rebuild in two sentences. +``` + +## Lab 5 — data agent instructions + +``` +agy -p "Read the table schemas: bq show --schema :cymbal_gold.fct_daily_revenue ; bq show --schema :cymbal_gold.dim_customer_360 ; bq show --schema :cymbal_gold.fct_product_performance. Then draft system instructions for a BigQuery conversational data agent over these three tables: synonyms business users might use (revenue, sales, LTV, best sellers), which table answers which kind of question, default groupings (order_date, currency, category), and columns to exclude from summaries (email). Output only the instructions text." +``` + +## Lab 6 — build the analyst agent + +Run inside `content/agenticdata/src/adk` (participants +`rm -rf cymbal_analyst` first so agy builds it from scratch). The full +specification lives in that folder's `AGENTS.md`: + +``` +/goal Read AGENTS.md, then create the files of the cymbal_analyst package +(cymbal_analyst/__init__.py, agent.py, a2a_server.py) exactly as it +describes. +``` + +Fallback: `git restore content/agenticdata/src/adk/cymbal_analyst/` + +## Lab 6 challenge — extend the concierge + +``` +/goal Add a second tool to cymbal_concierge/agent.py: +recent_orders(customer_email: str) returning the last 5 orders (id, status, +total, order_ts) for that customer from the live Postgres database, reusing +the existing connection pattern. Update the agent instruction so it knows +when to use which tool. +```