1. Introduzione
Questo codelab fornisce un progetto tecnico per i professionisti dei dati. Descrive un approccio "code-first" alla governance dei dati, mostrando come incorporare una solida gestione della qualità e dei metadati direttamente nel ciclo di vita dello sviluppo. Nella sua essenza, Knowledge Catalog funge da data fabric intelligente, consentendo alle organizzazioni di gestire, monitorare e governare centralmente i dati in tutto il loro patrimonio, dai data lake ai data warehouse.
Il codelab mostra come sfruttare Knowledge Catalog, BigQuery e la CLI Antigravity per appiattire dati complessi, profilarli in modo programmatico, generare suggerimenti intelligenti per le regole di qualità dei dati ed eseguire il deployment di scansioni di qualità automatizzate. L'obiettivo principale è andare oltre i processi manuali basati sulla UI, che sono soggetti a errori e difficili da scalare, e stabilire invece un framework "policy-as-code" solido e controllabile a livello di versione.
Obiettivi didattici
- Come appiattire i dati BigQuery nidificati utilizzando le visualizzazioni materializzate per consentire una profilazione completa.
- Come attivare e gestire programmaticamente le scansioni dei profili di Knowledge Catalog utilizzando la libreria client Python di Knowledge Catalog.
- Come esportare i dati del profilo e strutturarli come input per un modello di AI generativa.
- Come progettare un prompt per Antigravity CLI per analizzare i dati del profilo e generare un file di regole YAML conforme a Knowledge Catalog.
- L'importanza di un processo interattivo human-in-the-loop (HITL) per la convalida delle configurazioni generate dall'AI.
- Come eseguire il deployment delle regole generate come scansione automatica della qualità dei dati.
Che cosa ti serve
- Un account Google Cloud e un progetto Google Cloud
- Un browser web come Chrome
Concetti chiave: i pilastri della qualità dei dati di Knowledge Catalog
Comprendere i componenti principali di Knowledge Catalog è essenziale per creare una strategia efficace per la qualità dei dati.
- Scansione del profilo di dati: un job di Knowledge Catalog che analizza i dati e genera metadati statistici, tra cui percentuali di valori nulli, conteggi di valori distinti e distribuzioni dei valori. Questa è la nostra fase di "scoperta" programmatica.
- Regole di qualità dei dati:istruzioni dichiarative che definiscono le condizioni che i dati devono soddisfare (ad es.
NonNullExpectation,SetExpectation,RangeExpectation). - AI generativa per il suggerimento di regole:utilizzo di un modello linguistico di grandi dimensioni (come Gemini) per analizzare un profilo di dati e suggerire regole di qualità dei dati pertinenti. In questo modo, il processo di definizione di un framework di qualità di base viene accelerato.
- Scansione della qualità dei dati:un job di Knowledge Catalog che convalida i dati in base a un insieme di regole predefinite o personalizzate.
- Governance programmatica:il tema centrale della gestione dei controlli di governance (come le regole di qualità) come codice (ad es. in file YAML e script Python). Ciò consente l'automazione, il controllo delle versioni e l'integrazione nelle pipeline CI/CD.
- Human-in-the-loop (HITL): il punto di controllo fondamentale per integrare competenze e supervisione umane in un workflow automatizzato. Per le configurazioni generate con l'AI, HITL è essenziale per convalidare la correttezza, la pertinenza aziendale e la sicurezza dei suggerimenti prima dell'implementazione.
2. Configurazione e requisiti
Avvia Cloud Shell
Sebbene Google Cloud possa essere gestito da remoto dal tuo laptop, in questo codelab utilizzerai Google Cloud Shell, un ambiente a riga di comando in esecuzione nel cloud.
Nella console Google Cloud, fai clic sull'icona di Cloud Shell nella barra degli strumenti in alto a destra:
Bastano pochi istanti per eseguire il provisioning e connettersi all'ambiente. Al termine, dovresti vedere un risultato simile a questo:
Questa macchina virtuale è caricata con tutti gli strumenti per sviluppatori di cui avrai bisogno. Offre una home directory permanente da 5 GB e viene eseguita su Google Cloud, migliorando notevolmente le prestazioni e l'autenticazione della rete. Tutto il lavoro in questo codelab può essere svolto all'interno di un browser. Non devi installare nulla.
Abilita le API richieste e configura l'ambiente
In Cloud Shell, assicurati che l'ID progetto sia configurato:
export PROJECT_ID=$(gcloud config get-value project)
gcloud config set project $PROJECT_ID
export LOCATION="us-central1"
export BQ_LOCATION="us"
export DATASET_ID="kc_dq_codelab"
export TABLE_ID="ga4_transactions"
gcloud services enable dataplex.googleapis.com \
bigquery.googleapis.com \
serviceusage.googleapis.com \
aiplatform.googleapis.com
Nell'esempio utilizziamo us (multiregionale) come località, poiché anche i dati di esempio pubblici che utilizzeremo si trovano in us (multiregionale). BigQuery richiede che i dati di origine e la tabella di destinazione di una query si trovino nella stessa località.
Crea un set di dati BigQuery dedicato
Crea un nuovo set di dati BigQuery per ospitare i dati di esempio e i risultati.
bq --location=us mk --dataset $PROJECT_ID:$DATASET_ID
Preparare i dati di esempio
Per questo codelab, utilizzerai un set di dati pubblico contenente dati e-commerce offuscati del Google Merchandise Store. Poiché i set di dati pubblici sono di sola lettura, devi creare una copia modificabile nel tuo set di dati. Il seguente comando bq crea una nuova tabella, ga4_transactions, nel set di dati kc_dq_codelab. Copia i dati di un solo giorno (2021-01-31) per garantire che le scansioni vengano eseguite rapidamente.
bq query \
--use_legacy_sql=false \
--destination_table=$PROJECT_ID:$DATASET_ID.$TABLE_ID \
--replace=true \
'SELECT * FROM `bigquery-public-data.ga4_obfuscated_sample_ecommerce.events_20210131`'
Configurare la directory demo
Per iniziare, clonerai un repository GitHub che contiene la struttura di cartelle e i file di supporto necessari per questo codelab.
# Perform a shallow clone to get only the latest repository structure without the full history
git clone --depth 1 --filter=blob:none --sparse https://github.com/GoogleCloudPlatform/devrel-demos.git
cd devrel-demos
# Specify and download only the folder we need for this lab
git sparse-checkout set data-analytics/programmatic-dq
cd data-analytics/programmatic-dq
Questa directory è ora la tua area di lavoro attiva. Tutti i file successivi verranno creati qui.
3. Rilevamento automatico dei dati con la profilazione di Knowledge Catalog
Il profiling dei dati di Knowledge Catalog è un potente strumento per scoprire automaticamente informazioni statistiche sui tuoi dati, come percentuali di valori nulli, unicità e distribuzioni dei valori. Questo processo è essenziale per comprendere la struttura e la qualità dei dati. Tuttavia, una limitazione nota della profilazione di Knowledge Catalog è l'impossibilità di ispezionare completamente i campi nidificati o ripetuti (ad es. tipi RECORD o ARRAY) all'interno di una tabella. Può identificare una colonna come tipo complesso, ma non può profilare i singoli campi all'interno di questa struttura nidificata.
Per superare questo problema, appiattiremo i dati in viste materializzate create appositamente. Questa strategia rende ogni campo una colonna di primo livello, consentendo a Knowledge Catalog di profilare ciascuno individualmente.
Informazioni sullo schema nidificato
Innanzitutto, esaminiamo lo schema della tabella di origine. Il set di dati Google Analytics 4 (GA4) contiene diverse colonne nidificate e ripetute. Per recuperare in modo programmatico lo schema completo, incluse tutte le strutture nidificate, puoi utilizzare il comando bq show e salvare l'output come file JSON.
bq show --schema --format=json $PROJECT_ID:$DATASET_ID.$TABLE_ID > bq_schema.json
L'ispezione del file bq_schema.json rivela strutture complesse come dispositivo, dati geografici, e-commerce e elementi di record ripetuti. Queste sono le strutture che richiedono l'appiattimento per una profilazione efficace.
Appiattimento dei dati con le viste materializzate
La creazione di viste materializzate (MV) è la soluzione più efficace e pratica a questa sfida dei dati nidificati. Calcolando in anticipo i risultati appiattiti, le MV offrono vantaggi significativi in termini di prestazioni e costi delle query, fornendo al contempo una struttura più semplice e simile a quella relazionale per gli analisti e gli strumenti di profilazione.
Il primo pensiero potrebbe essere quello di appiattire tutto in un'unica visualizzazione gigante. Tuttavia, questo approccio intuitivo nasconde una trappola pericolosa che può portare a una grave corruzione dei dati. Vediamo perché si tratta di un errore critico.
mv_ga4_user_session_flat.sql
CREATE OR REPLACE MATERIALIZED VIEW `$PROJECT_ID.$DATASET_ID.mv_ga4_user_session_flat`
OPTIONS (
enable_refresh = true,
refresh_interval_minutes = 30
) AS
SELECT
event_date, event_timestamp, event_name, user_pseudo_id, user_id, stream_id, platform,
device.category AS device_category,
device.operating_system AS device_os,
device.operating_system_version AS device_os_version,
device.language AS device_language,
device.web_info.browser AS device_browser,
geo.continent AS geo_continent,
geo.country AS geo_country,
geo.region AS geo_region,
geo.city AS geo_city,
traffic_source.name AS traffic_source_name,
traffic_source.medium AS traffic_source_medium,
traffic_source.source AS traffic_source_source
FROM
`$PROJECT_ID.$DATASET_ID.ga4_transactions`;
mv_ga4_ecommerce_transactions.sql
CREATE OR REPLACE MATERIALIZED VIEW `$PROJECT_ID.$DATASET_ID.mv_ga4_ecommerce_transactions`
OPTIONS (
enable_refresh = true,
refresh_interval_minutes = 30
) AS
SELECT
event_date, event_timestamp, user_pseudo_id, ecommerce.transaction_id,
ecommerce.total_item_quantity,
ecommerce.purchase_revenue_in_usd,
ecommerce.purchase_revenue,
ecommerce.refund_value_in_usd,
ecommerce.refund_value,
ecommerce.shipping_value_in_usd,
ecommerce.shipping_value,
ecommerce.tax_value_in_usd,
ecommerce.tax_value,
ecommerce.unique_items
FROM
`$PROJECT_ID.$DATASET_ID.ga4_transactions`
WHERE
ecommerce.transaction_id IS NOT NULL;
mv_ga4_ecommerce_items.sql
CREATE OR REPLACE MATERIALIZED VIEW `$PROJECT_ID.$DATASET_ID.mv_ga4_ecommerce_items`
OPTIONS (
enable_refresh = true,
refresh_interval_minutes = 30
) AS
SELECT
event_date, event_timestamp, event_name, user_pseudo_id, ecommerce.transaction_id,
item.item_id,
item.item_name,
item.item_brand,
item.item_variant,
item.item_category,
item.item_category2,
item.item_category3,
item.item_category4,
item.item_category5,
item.price_in_usd,
item.price,
item.quantity,
item.item_revenue_in_usd,
item.item_revenue,
item.coupon,
item.affiliation,
item.item_list_name,
item.promotion_name
FROM
`$PROJECT_ID.$DATASET_ID.ga4_transactions`,
UNNEST(items) AS item
WHERE
ecommerce.transaction_id IS NOT NULL;
Ora esegui questi modelli utilizzando lo strumento a riga di comando bq. Il comando envsubst leggerà ogni file, sostituirà le variabili come $PROJECT_ID e $DATASET_ID con i relativi valori dell'ambiente shell e invierà l'SQL finale e valido al comando bq query.
envsubst < mv_ga4_user_session_flat.sql | bq query --use_legacy_sql=false
envsubst < mv_ga4_ecommerce_transactions.sql | bq query --use_legacy_sql=false
envsubst < mv_ga4_ecommerce_items.sql | bq query --use_legacy_sql=false
Esegui le scansioni dei profili tramite il client Python
Ora che abbiamo le visualizzazioni profilabili e compresse, possiamo creare ed eseguire in modo programmatico le scansioni di profilazione dei dati di Knowledge Catalog per ciascuna. Il seguente script Python utilizza la libreria client google-cloud-dataplex per automatizzare questo processo.
Prima di eseguire lo script, è una best practice fondamentale creare un ambiente Python isolato all'interno della directory del progetto. In questo modo, le dipendenze del progetto vengono gestite separatamente, evitando conflitti con altri pacchetti nell'ambiente Cloud Shell.
# Create the virtual environment
python3 -m venv dq_venv
# Activate the environment
source dq_venv/bin/activate
Ora installa la libreria client Knowledge Catalog all'interno dell'ambiente appena attivato.
# Install the Knowledge Catalog client library
pip install google-cloud-dataplex
Con l'ambiente configurato e la libreria installata, puoi creare lo script di orchestrazione.
Nella barra degli strumenti di Cloud Shell, fai clic su Apri editor. Crea un nuovo file denominato 1_run_scan.py e incolla il seguente codice Python. Se cloni il repository GitHub, questo file è già nella tua cartella.
Questo script creerà una scansione per ogni vista materializzata (se non esiste già), eseguirà la scansione e poi eseguirà il polling finché tutti i job di scansione non saranno completati.
import os
import sys
import time
from google.cloud import dataplex_v1
from google.api_core.exceptions import AlreadyExists
def create_and_run_scan(
client: dataplex_v1.DataScanServiceClient,
project_id: str,
location: str,
data_scan_id: str,
target_resource: str,
) -> dataplex_v1.DataScanJob | None:
"""
Creates and runs a single data profile scan.
Returns the executed Job object without waiting for completion.
"""
parent = client.data_scan_path(project_id, location, data_scan_id).rsplit('/', 2)[0]
scan_path = client.data_scan_path(project_id, location, data_scan_id)
# 1. Create Data Scan (skips if it already exists)
try:
data_scan = dataplex_v1.DataScan()
data_scan.data.resource = target_resource
data_scan.data_profile_spec = dataplex_v1.DataProfileSpec()
print(f"[INFO] Creating data scan '{data_scan_id}'...")
client.create_data_scan(
parent=parent,
data_scan=data_scan,
data_scan_id=data_scan_id
).result() # Wait for creation to complete
print(f"[SUCCESS] Data scan '{data_scan_id}' created.")
except AlreadyExists:
print(f"[INFO] Data scan '{data_scan_id}' already exists. Skipping creation.")
except Exception as e:
print(f"[ERROR] Error creating data scan '{data_scan_id}': {e}")
return None
# 2. Run Data Scan
try:
print(f"[INFO] Running data scan '{data_scan_id}'...")
run_response = client.run_data_scan(name=scan_path)
print(f"[SUCCESS] Job started for '{data_scan_id}'. Job ID: {run_response.job.name.split('/')[-1]}")
return run_response.job
except Exception as e:
print(f"[ERROR] Error running data scan '{data_scan_id}': {e}")
return None
def main():
"""Main execution function"""
# --- Load configuration from environment variables ---
PROJECT_ID = os.environ.get("PROJECT_ID")
LOCATION = os.environ.get("LOCATION")
DATASET_ID = os.environ.get("DATASET_ID")
if not all([PROJECT_ID, LOCATION, DATASET_ID]):
print("[ERROR] One or more required environment variables are not set.")
print("Please ensure PROJECT_ID, LOCATION, and DATASET_ID are exported in your shell.")
sys.exit(1)
print(f"[INFO] Using Project: {PROJECT_ID}, Location: {LOCATION}, Dataset: {DATASET_ID}")
# List of Materialized Views to profile
TARGET_VIEWS = [
"mv_ga4_user_session_flat",
"mv_ga4_ecommerce_transactions",
"mv_ga4_ecommerce_items"
]
# ----------------------------------------------------
client = dataplex_v1.DataScanServiceClient()
running_jobs = []
# 1. Create and run jobs for all target views
print("\n--- Starting Data Profiling Job Creation and Execution ---")
for view_name in TARGET_VIEWS:
data_scan_id = f"profile-scan-{view_name.replace('_', '-')}"
target_resource = f"//bigquery.googleapis.com/projects/{PROJECT_ID}/datasets/{DATASET_ID}/tables/{view_name}"
job = create_and_run_scan(client, PROJECT_ID, LOCATION, data_scan_id, target_resource)
if job:
running_jobs.append(job)
print("-------------------------------------------------------\n")
if not running_jobs:
print("[ERROR] No jobs were started. Exiting.")
return
# 2. Poll for all jobs to complete
print("--- Monitoring job completion status (checking every 30 seconds) ---")
completed_jobs = {}
while running_jobs:
jobs_to_poll_next = []
print(f"\n[STATUS] Checking status for {len(running_jobs)} running jobs...")
for job in running_jobs:
job_id_short = job.name.split('/')[-1][:13]
try:
updated_job = client.get_data_scan_job(name=job.name)
state = updated_job.state
if state in (dataplex_v1.DataScanJob.State.RUNNING, dataplex_v1.DataScanJob.State.PENDING, dataplex_v1.DataScanJob.State.CANCELING):
print(f" - Job {job_id_short}... Status: {state.name}")
jobs_to_poll_next.append(updated_job)
else:
print(f" - Job {job_id_short}... Status: {state.name} (Complete)")
completed_jobs[job.name] = updated_job
except Exception as e:
print(f"[ERROR] Could not check status for job {job_id_short}: {e}")
running_jobs = jobs_to_poll_next
if running_jobs:
time.sleep(30)
# 3. Print final results
print("\n--------------------------------------------------")
print("[SUCCESS] All data profiling jobs have completed.")
print("\nFinal Job Status Summary:")
for job_name, job in completed_jobs.items():
job_id_short = job_name.split('/')[-1][:13]
print(f" - Job {job_id_short}: {job.state.name}")
if job.state == dataplex_v1.DataScanJob.State.FAILED:
print(f" - Failure Message: {job.message}")
print("\nNext step: Analyze the profile results and generate quality rules.")
if __name__ == "__main__":
main()
Ora esegui lo script dal terminale Cloud Shell.
python 1_run_scan.py
Lo script ora orchestra la profilazione delle tre viste materializzate, fornendo aggiornamenti di stato in tempo reale. Al termine, avrai un profilo statistico ricco e leggibile automaticamente per ogni visualizzazione, pronto per la fase successiva del nostro flusso di lavoro: la generazione di regole di qualità dei dati basate sull'AI.
Puoi visualizzare le scansioni dei profili completate nella console Google Cloud.
- Nel menu di navigazione, vai a Knowledge Catalog e Profilazione e qualità dei dati nella sezione Governa.
- Dovresti visualizzare i tre profili scansionati elencati, insieme al loro stato del job più recente. Puoi fare clic su una scansione per visualizzare i risultati dettagliati.
Dal profilo BigQuery all'input pronto per l'AI
Le scansioni dei profili di Knowledge Catalog sono state eseguite correttamente. Sebbene i risultati siano disponibili nell'API Knowledge Catalog, per utilizzarli come input per un modello di AI generativa, dobbiamo estrarli in un file locale strutturato.
Il seguente script Python, 2_dq_profile_save.py, trova a livello di programmazione l'ultimo job di scansione di profilazione riuscito per la nostra visualizzazione mv_ga4_user_session_flat. Recupera quindi il risultato del profilo completo e dettagliato e lo salva come file JSON locale denominato dq_profile_results.json. Questo file fungerà da input diretto per la nostra analisi AI nel passaggio successivo.
Nell'editor di Cloud Shell, crea un nuovo file denominato 2_dq_profile_save.py e incolla il seguente codice. Come nel passaggio precedente, puoi saltare la creazione di un file se hai clonato il repository.
import os
import sys
import json
from google.cloud import dataplex_v1
from google.api_core.exceptions import NotFound
from google.protobuf.json_format import MessageToDict
# --- Configuration ---
# The Materialized View to analyze is fixed for this step.
TARGET_VIEW = "mv_ga4_user_session_flat"
OUTPUT_FILENAME = "dq_profile_results.json"
def save_to_json_file(content: dict, filename: str):
"""Saves the given dictionary content to a JSON file."""
try:
with open(filename, "w", encoding="utf-8") as f:
# Use indent=2 for a readable, "pretty-printed" JSON file.
json.dump(content, f, indent=2, ensure_ascii=False)
print(f"\n[SUCCESS] Profile results were saved to '{filename}'.")
except (IOError, TypeError) as e:
print(f"[ERROR] An error occurred while saving the file: {e}")
def get_latest_successful_job(
client: dataplex_v1.DataScanServiceClient,
project_id: str,
location: str,
data_scan_id: str
) -> dataplex_v1.DataScanJob | None:
"""Finds and returns the most recently succeeded job for a given data scan."""
scan_path = client.data_scan_path(project_id, location, data_scan_id)
print(f"\n[INFO] Looking for the latest successful job for scan '{data_scan_id}'...")
try:
# List all jobs for the specified scan, which are ordered most-recent first.
jobs_pager = client.list_data_scan_jobs(parent=scan_path)
# Iterate through jobs to find the first one that succeeded.
for job in jobs_pager:
if job.state == dataplex_v1.DataScanJob.State.SUCCEEDED:
return job
# If no successful job is found after checking all pages.
return None
except NotFound:
print(f"[WARN] No scan history found for '{data_scan_id}'.")
return None
def main():
"""Main execution function."""
# --- Load configuration from environment variables ---
PROJECT_ID = os.environ.get("PROJECT_ID")
LOCATION = os.environ.get("LOCATION")
if not all([PROJECT_ID, LOCATION]):
print("[ERROR] Required environment variables PROJECT_ID or LOCATION are not set.")
sys.exit(1)
print(f"[INFO] Using Project: {PROJECT_ID}, Location: {LOCATION}")
print(f"--- Starting Profile Retrieval for: {TARGET_VIEW} ---")
# Construct the data_scan_id based on the target view name.
data_scan_id = f"profile-scan-{TARGET_VIEW.replace('_', '-')}"
# 1. Initialize client and get the latest successful job.
client = dataplex_v1.DataScanServiceClient()
latest_job = get_latest_successful_job(client, PROJECT_ID, LOCATION, data_scan_id)
if not latest_job:
print(f"\n[ERROR] No successful job record was found for '{data_scan_id}'.")
print("Please ensure the '1_run_scan.py' script has completed successfully.")
return
job_id_short = latest_job.name.split('/')[-1]
print(f"[SUCCESS] Found the latest successful job: '{job_id_short}'.")
# 2. Fetch the full, detailed profile result for the job.
print(f"[INFO] Retrieving detailed profile results for job '{job_id_short}'...")
try:
request = dataplex_v1.GetDataScanJobRequest(
name=latest_job.name,
view=dataplex_v1.GetDataScanJobRequest.DataScanJobView.FULL,
)
job_with_full_results = client.get_data_scan_job(request=request)
except Exception as e:
print(f"[ERROR] Failed to retrieve detailed job results: {e}")
return
# 3. Convert the profile result to a dictionary and save it to a JSON file.
if job_with_full_results.data_profile_result:
profile_dict = MessageToDict(job_with_full_results.data_profile_result._pb)
save_to_json_file(profile_dict, OUTPUT_FILENAME)
else:
print("[WARN] The job completed, but no data profile result was found within it.")
print("\n[INFO] Script finished successfully.")
if __name__ == "__main__":
main()
Ora esegui lo script dal terminale:
python 2_dq_profile_save.py
Al termine dell'operazione, nella directory sarà presente un nuovo file denominato dq_profile_results.json. Questo file contiene i metadati statistici dettagliati e avanzati che utilizzeremo per generare regole di qualità dei dati. Se vuoi controllare i contenuti di dq_profile_results.json, esegui il seguente comando:
cat dq_profile_results.json
4. Generare regole per la qualità dei dati con l'interfaccia a riga di comando Antigravity
Ora puoi utilizzare Antigravity CLI (agy) per leggere i risultati della scansione di profilazione locale. Scrivere manualmente le specifiche di qualità dei dati per set di dati complessi richiede molto tempo ed è soggetto a errori. L'utilizzo di un agente AI generativa all'interno del terminale accelera questo flusso di lavoro creando una configurazione dichiarativa iniziale in pochi secondi. In questo modo, i team di dati possono passare dalla stesura manuale della sintassi alla supervisione di Human-in-the-Loop (HITL) di alto livello e in linea con l'attività.
Per avviare Antigravity CLI, esegui questo comando nel terminale:
agy
Ora puoi generare regole di qualità. Poiché la CLI può leggere i file nella directory corrente, può utilizzare direttamente i nuovi dati di scansione di profilazione.
Chiedere all'agente di creare un piano
Innanzitutto, chiederemo all'agente Antigravity di analizzare il profilo statistico e proporre un piano d'azione. Gli chiediamo esplicitamente di non scrivere ancora il file YAML. In questo modo, l'attenzione si concentra sull'analisi e sulla giustificazione.
Nella sessione interattiva di Antigravity CLI, inserisci il seguente prompt strutturato:
# Context
You are preparing a data quality rule configuration plan for Google Cloud Knowledge Catalog based on data profile statistics.
# Input
- File Path: `./dq_profile_results.json` (contains metrics like null percentage, distinct counts, and distributions)
# Task
Analyze the input statistics and propose a step-by-step plan for establishing automated data quality rules.
*Do not write any YAML code in this step.* Focus only on analytical planning.
# Rule Mapping Strategy
For candidate columns, match the statistical metrics to the most appropriate expectations:
- `nonNullExpectation`: Propose for columns with 0% null values in the profile.
- `setExpectation`: Propose for columns with a highly limited, stable set of categorical values.
- `rangeExpectation`: Propose for numeric columns with consistent and predictable value boundaries.
# Guidelines
- Provide a metric-based justification for each proposed rule (e.g., "Recommend `nonNullExpectation` for column 'user_pseudo_id' because its null percentage is 0%").
- Flag volatile metrics such as hardcoded row counts that could cause false-positive alerts in production.
# Output Format
Provide your analysis and proposed rules as a structured, step-by-step markdown plan with clear headings.
L'agente analizzerà il file JSON e dovrebbe restituire un piano strutturato simile a questo:
Automated Data Quality Rule Configuration Plan
Google Cloud Knowledge Catalog (Dataplex Data Quality)
──────
## Executive Summary
This analytical planning document outlines a step-by-step strategy for configuring automated data quality (DQ) rules in Google Cloud Knowledge Catalog (formerly Dataplex Data Quality) based on profiling statistics.
The dataset contains 26,489 rows representing GA4 event logs. Based on statistical metrics (null ratios, distinct value distributions, and data types), candidate columns are mapped to appropriate expectation rules.
──────
## 1. Data Profile Overview & Statistical Highlights
Column Name │ Data Type │ Null Ratio │ Distinct Count │ Key Value Range / Categories
─────────────────┼───────────┼────────────────┼────────────────┼──────────────────────────────────────────────────
event_date │ STRING │ 0.0% (0) │ 1 (3.78e-05) │ "20210131" (100%)
event_timestamp │ INTEGER │ 0.0% (0) │ ~16,539 (0.62) │ Min: 1612051200657906, Max: 1612137595412363
event_name │ STRING │ 0.0% (0) │ 16 (0.0006) │ page_view (35.8%), user_engagement (18.9%), etc.
user_pseudo_id │ STRING │ 0.0% (0) │ ~2,545 (0.09) │ 18–21 characters string identifiers
user_id │ STRING │ 100.0% (1.0) │ 0 (0.0) │ Entirely NULL
device_category │ STRING │ 0.0% (0) │ 3 (0.0001) │ desktop (57.5%), mobile (40.1%), tablet (2.4%)
... │ ... │ ... │ ... │ ...
──────
## 2. Rule Mapping Strategy & Analytical Justifications
### Step 1: Nullability Rules (nonNullExpectation)
Propose nonNullExpectation for mandatory columns where the data profile demonstrates 0% null values.
• user_pseudo_id, event_timestamp, event_name, event_date, stream_id, platform, device_category (Metric Justification: nullRatio is 0.0%)
│ [!NOTE] Exclusions:
│ • user_id: Has a nullRatio of 100.0% (unauthenticated traffic).
│ • device_language: Has a nullRatio of 37.53%.
──────
### Step 2: Categorical Value Set Validation (setExpectation)
Propose setExpectation for columns with a highly limited, stable set of categorical domain values.
• device_category: Distinct count is exactly 3. Allowed set: ['desktop', 'mobile', 'tablet']
• platform: Distinct count is 1. Allowed set expanded to: ['WEB', 'ANDROID', 'IOS'] to avoid over-fitting.
• geo_continent: Distinct count is 6. Allowed set: ['Americas', 'Asia', 'Europe', 'Africa', 'Oceania', 'Antarctica', '(not set)']
──────
### Step 3: Numeric & Timestamp Boundary Validation (rangeExpectation)
Propose rangeExpectation for numeric columns with consistent and predictable value boundaries.
• event_timestamp: rangeExpectation requiring event_timestamp > 0 (avoid dynamic microsecond range hardcoding)
• stream_id: rangeExpectation requiring positive integer stream IDs (stream_id > 0)
──────
## 3. Risk Warning: Volatile Metrics & Production False Positives
│ [!WARNING] Volatile Metrics Flagged for Risk Mitigation:
1. Hardcoded Total Row Count (rowCount = 26,489) -> Daily event volume fluctuates. Use dynamic volume thresholds.
2. Hardcoded Partition Date (event_date = '20210131') -> Breaks on future runs. Validate against YYYYMMDD regex patterns.
3. Exact Timestamp Range Bounds -> Enforcing these microsecond limits on incoming live pipelines will reject all future data.
4. Single-Value Domain Restrictions -> Single profile sample might lack active streams. Set sets according to enterprise schema.
──────
## Summary Table of Proposed Rules
Target Column │ Rule Type │ Metric-Based Justification │ Operational Considerations
─────────────────┼────────────────────┼────────────────────────────┼──────────────────────────────────────────────────
user_pseudo_id │ nonNullExpectation │ Null Ratio: 0.0% │ Core identifier, strictly required
event_timestamp │ nonNullExpectation │ Null Ratio: 0.0% │ Temporal key, strictly required
event_timestamp │ rangeExpectation │ Min: > 0 (Microseconds) │ Avoid hardcoding epoch min/max
event_name │ nonNullExpectation │ Null Ratio: 0.0% │ Required event taxonomy key
event_name │ setExpectation │ Categorical distribution │ Map to standard GA4 event taxonomy
device_category │ nonNullExpectation │ Null Ratio: 0.0% │ Required form-factor dimension
device_category │ setExpectation │ Distinct Count: 3 values │ ['desktop', 'mobile', 'tablet']
... │ ... │ ... │ ...
Genera le regole sulla qualità dei dati
Questo è il passaggio più importante dell'intero flusso di lavoro: la revisione human-in-the-loop (HITL). Il piano generato dall'agente si basa esclusivamente su pattern statistici nei dati. Non ha alcuna comprensione del contesto aziendale, delle modifiche future ai dati o dell'intento specifico alla base dei dati. Il tuo ruolo di esperto umano è quello di convalidare, correggere e approvare questo piano prima di trasformarlo in codice.
Esamina attentamente il piano fornito dall'agente.
- Ha senso?
- È in linea con le tue conoscenze aziendali?
- Esistono regole statisticamente valide ma praticamente inutili?
L'output che ricevi dall'agente può variare. Il tuo obiettivo è perfezionarlo. Ad esempio, immagina che il piano suggerisca una regola rowCount perché la tabella ha un numero fisso di righe nei dati di esempio. In qualità di esperto umano, potresti sapere che le dimensioni di questa tabella aumentano ogni giorno, il che rende impraticabile una regola rigida sul conteggio delle righe e potrebbe causare falsi avvisi. Questo è un esempio perfetto di applicazione del contesto aziendale che l'AI non ha.
Ora fornisci un feedback all'agente e dai il comando finale per generare il codice. Devi adattare il seguente prompt in base al piano che hai effettivamente ricevuto e alle correzioni che vuoi apportare.
Il prompt riportato di seguito è un modello. La prima riga è quella in cui fornirai le correzioni specifiche. Se il piano che ti ha fornito l'agente è perfetto e non necessita di modifiche, puoi semplicemente eliminare la riga.
Nella stessa sessione di Antigravity, inserisci la tua versione adattata del seguente prompt:
# Feedback & Approvals
[YOUR CORRECTIONS AND APPROVAL GO HERE. Examples:
- "The plan looks good. Please proceed."
- "The rowCount rule is not necessary, as the table size changes daily. The rest of the plan is approved. Please proceed."
- "For the setExpectation on the geo_continent column, please also include 'Antarctica'."]
# Objective
Based on the approved analysis plan and the provided feedback, generate the final `dq_rules.yaml` file conforming to the standard `DataQualityRule` schema.
# Instructions
1. **Rule Justifications**: For every generated rule, add a YAML comment (`#`) on the line directly above it, briefly explaining the justification established in the plan.
2. **Schema Alignment**: Ensure the structure strictly adheres to the required Knowledge Catalog data quality scan specification. Refer to the `sample_rule.yaml` file in the current directory and the `DataQualityRule` class definition in the local virtual environment path (`./dq_venv/.../google/cloud/dataplex_v1/types/data_quality.py`) as the schema authority.
3. **Data-Driven Values**: Derive all rule parameters, such as thresholds or expected values, directly from the statistical metrics in `dq_profile_results.json`.
# Constraints
- **Output Purity**: Return ONLY the raw, valid, and properly formatted YAML code block.
- Do not include conversational preambles, introductory sentences, explanations, or markdown blocks around the YAML.
L'agente genererà ora i contenuti YAML in base alle tue istruzioni precise e convalidati da persone. Al termine, nella directory di lavoro troverai un nuovo file denominato dq_rules.yaml.
Crea ed esegui la scansione della qualità dei dati
Ora che hai un file dq_rules.yaml generato dall'agente e con la convalida umana, puoi eseguirne il deployment in tutta sicurezza.
Esci dalla CLI di Antigravity digitando /quit o premendo due volte Ctrl+C.
Il seguente comando gcloud crea una nuova risorsa di scansione dei dati di Knowledge Catalog. Non esegue ancora la scansione, ma registra semplicemente la definizione e la configurazione della scansione (il nostro file YAML) in Knowledge Catalog.
Esegui questo comando nel terminale:
export DQ_SCAN="dq-scan"
gcloud dataplex datascans create data-quality $DQ_SCAN \
--project=$PROJECT_ID \
--location=$LOCATION \
--data-quality-spec-file=dq_rules.yaml \
--data-source-resource="//bigquery.googleapis.com/projects/$PROJECT_ID/datasets/$DATASET_ID/tables/mv_ga4_user_session_flat"
Ora che la scansione è definita, puoi attivare un job per eseguirla.
gcloud dataplex datascans run $DQ_SCAN --location=$LOCATION --project=$PROJECT_ID
Questo comando restituirà un ID job. Puoi monitorare lo stato di questo job nella sezione Knowledge Catalog della console Google Cloud. Al termine, i risultati verranno scritti in una tabella BigQuery per l'analisi.
5. Il ruolo fondamentale di Human-In-The-Loop (HITL)
Sebbene l'utilizzo dell'agente Antigravity per accelerare la generazione di regole sia incredibilmente efficace, è fondamentale trattare l'AI come un copilota altamente qualificato, non come un pilota completamente autonomo. Il processo human-in-the-loop (HITL) non è un suggerimento facoltativo, ma un passaggio fondamentale e non negoziabile in qualsiasi flusso di lavoro di governance dei dati. Il semplice deployment di artefatti generati dall'AI senza una rigorosa supervisione umana è una ricetta per il fallimento.
Considera la dq_rules.yaml generata dall'AI come una richiesta di pull inviata da uno sviluppatore di AI estremamente veloce ma inesperto. Richiede una revisione approfondita da parte di un esperto umano senior, ovvero te, prima di poter essere unito al "ramo principale" della tua norma di governance e implementato. Questa revisione è essenziale per mitigare i punti deboli intrinseci dei modelli linguistici di grandi dimensioni.
Ecco una suddivisione dettagliata del motivo per cui questa revisione umana è indispensabile e di cosa devi cercare in particolare:
1. Convalida contestuale: l'AI non è consapevole dell'attività
- Il punto debole dell'LLM: un LLM è un maestro di pattern e statistiche, ma non ha alcuna comprensione del contesto aziendale. Ad esempio, se una colonna,
new_campaign_id, ha un rapporto di valori null del 98%, un LLM potrebbe ignorarla per un motivo statistico. - Il ruolo fondamentale dell'essere umano: tu, l'esperto umano, sai che il campo
new_campaign_idè stato aggiunto solo ieri per il lancio di un prodotto importante la prossima settimana. Sai che il suo rapporto di valori nulli dovrebbe essere elevato ora, ma è previsto un calo significativo. Sai anche che, una volta compilato, deve seguire un formato specifico. L'AI non può dedurre queste conoscenze aziendali esterne. Il tuo ruolo è quello di applicare questo contesto aziendale ai suggerimenti statistici dell'AI, ignorandoli o aumentandoli se necessario.
2. Correttezza e precisione: protezione da allucinazioni ed errori sottili
- Il punto debole degli LLM:gli LLM possono essere "sicuri di sbagliare". Possono "avere allucinazioni" o generare codice leggermente errato. Ad esempio, potrebbe generare un file YAML con una regola denominata correttamente, ma con un parametro non valido, oppure potrebbe scrivere in modo errato un tipo di regola (ad es.
setExpectationsanzichésetExpectation). Questi errori sottili causeranno l'esito negativo del deployment, ma possono essere difficili da individuare. - Il ruolo fondamentale dell'essere umano: il tuo compito è fungere da linter e strumento di convalida di schema definitivo. Devi controllare meticolosamente il file YAML generato rispetto alla specifica
DataQualityRuleufficiale di Knowledge Catalog. Non stai solo controllando se "sembra giusto", ma stai convalidando la correttezza sintattica e semantica per assicurarti che sia conforme al 100% all'API di destinazione. Per questo motivo, il codelab chiede all'agente di fare riferimento ai file di schema, per ridurre la possibilità di errori, ma la verifica finale spetta a te.
3. Sicurezza e mitigazione dei rischi: prevenzione delle conseguenze a valle
- Il punto debole del LLM:una regola sulla qualità dei dati difettosa implementata in produzione può avere gravi conseguenze. Se l'AI suggerisce un
rangeExpectationper un importo di transazione finanziaria troppo ampio, potrebbe non rilevare attività fraudolente. Al contrario, se suggerisce una regola troppo rigida basata su un piccolo campione di dati, potrebbe inondare il tuo team di reperibilità con migliaia di avvisi di falsi positivi, causando affaticamento da avvisi e facendo sì che i problemi reali vengano ignorati. - Il ruolo fondamentale dell'essere umano: sei l'ingegnere della sicurezza. Devi valutare il potenziale impatto a valle di ogni singola regola suggerita dall'AI. Chiediti: "Cosa succede se questa regola non funziona? L'avviso può essere seguito da un'azione immediata? Qual è il rischio se questa regola viene superata in modo errato?" Questa valutazione del rischio è una capacità unicamente umana che mette a confronto il costo del fallimento con il vantaggio del controllo.
4. Governance come processo continuo: incorporare conoscenze lungimiranti
- Il punto debole del LLM:le conoscenze dell'AI si basano su un'istantanea statica dei dati. I risultati del profilo si riferiscono a un momento specifico. Non ha informazioni sugli eventi futuri.
- Il ruolo fondamentale dell'essere umano: la tua strategia di governance deve essere lungimirante. Sai che la migrazione di un'origine dati è pianificata per il mese successivo, il che cambierà lo stream_id. Sai che un nuovo paese viene aggiunto all'elenco
geo_country. La procedura HITL prevede l'inserimento di queste informazioni sullo stato futuro, l'aggiornamento o la disattivazione temporanea delle regole per evitare interruzioni durante le evoluzioni tecniche o aziendali pianificate. La qualità dei dati non è una configurazione una tantum, ma un processo dinamico che deve evolversi e solo un essere umano può guidare questa evoluzione.
In sintesi, l'intervento umano nel ciclo di vita dell'AI è il meccanismo essenziale di garanzia di qualità e di sicurezza che trasforma la governance basata sull'AI da un'idea nuova ma rischiosa in una pratica responsabile, scalabile e di livello enterprise. Garantisce che le policy finali implementate non siano solo accelerate dall'AI, ma anche convalidate da persone, combinando la velocità delle macchine con la saggezza e il contesto degli esperti umani.
Tuttavia, questa enfasi sulla supervisione umana non diminuisce il valore dell'AI. Al contrario, l'AI generativa svolge un ruolo cruciale nell'accelerare il processo HITL stesso.
Senza l'AI, un data engineer dovrebbe:
- Scrivi manualmente query SQL complesse per profilare i dati (ad es.
COUNT DISTINCT,AVG,MIN,MAXper ogni colonna). - Analizza meticolosamente i risultati foglio di lavoro per foglio di lavoro.
- Scrivere ogni singola riga del file di regole YAML da zero, un'attività noiosa e soggetta a errori.
L'AI automatizza questi passaggi laboriosi e dispendiosi in termini di tempo. Agisce come un analista instancabile che elabora istantaneamente il profilo statistico e fornisce una "prima bozza" della norma ben strutturata e completa all'80%.
Ciò cambia radicalmente la natura del lavoro dell'essere umano. Anziché dedicare ore all'elaborazione manuale dei dati e alla codifica boilerplate, l'esperto umano può concentrarsi immediatamente sulle attività di maggior valore:
- Applicazione del contesto aziendale critico.
- Convalida della correttezza della logica dell'AI.
- Prendere decisioni strategiche su quali regole sono davvero importanti.
In questa partnership, l'AI gestisce il "cosa" (quali sono i pattern statistici?), lasciando all'essere umano la possibilità di concentrarsi sul "perché" (perché questo pattern è importante per la nostra attività?) e sul "quindi" (quindi quale dovrebbe essere la nostra policy?). Pertanto, l'AI non sostituisce il ciclo, ma rende ogni ciclo più veloce, intelligente e di maggiore impatto.
6. Pulizia dell'ambiente
Per evitare che al tuo account Google Cloud vengano addebitati costi futuri per le risorse utilizzate in questo codelab, devi eliminare il progetto che contiene le risorse. Tuttavia, se vuoi conservare il progetto, puoi eliminare le singole risorse che hai creato.
Elimina le scansioni di Knowledge Catalog
Innanzitutto, elimina le scansioni del profilo e della qualità che hai creato. Per evitare l'eliminazione accidentale di risorse importanti, questi comandi utilizzano i nomi specifici delle scansioni create in questo codelab.
# Delete the Data Quality Scan
gcloud dataplex datascans delete dq-scan \
--location=us-central1 \
--project=$PROJECT_ID --quiet
# Delete the Data Profile Scans
gcloud dataplex datascans delete profile-scan-mv-ga4-user-session-flat \
--location=us-central1 \
--project=$PROJECT_ID --quiet
gcloud dataplex datascans delete profile-scan-mv-ga4-ecommerce-transactions \
--location=us-central1 \
--project=$PROJECT_ID --quiet
gcloud dataplex datascans delete profile-scan-mv-ga4-ecommerce-items \
--location=us-central1 \
--project=$PROJECT_ID --quiet
Elimina il set di dati BigQuery
Poi, elimina il set di dati BigQuery. Questo comando è irreversibile e utilizza il flag -f (forza) per rimuovere il set di dati e tutte le relative tabelle senza conferma.
# Manually type this command to confirm you are deleting the correct dataset
bq rm -r -f --dataset $PROJECT_ID:kc_dq_codelab
7. Complimenti!
Hai completato il codelab.
Hai creato un workflow di governance dei dati end-to-end e programmatico. Hai iniziato utilizzando le viste materializzate per appiattire i dati BigQuery complessi, rendendoli adatti all'analisi. Successivamente, hai eseguito in modo programmatico le scansioni dei profili di Knowledge Catalog per generare metadati statistici. Ancora più importante, hai utilizzato la Antigravity CLI per analizzare l'output del profilo e generare in modo intelligente un artefatto "policy-as-code" (dq_rules.yaml). Poi hai utilizzato la CLI per fare il deployment di questa configurazione come scansione automatica della qualità dei dati, chiudendo il cerchio di una strategia di governance moderna e scalabile.
Ora disponi del pattern fondamentale per creare sistemi di qualità dei dati affidabili, accelerati dall'AI e con convalida umana su Google Cloud.
Passaggi successivi
- Integra con CI/CD:prendi il file
dq_rules.yamle eseguine il commit in un repository Git. Crea una pipeline CI/CD (ad es. utilizzando Cloud Build o GitHub Actions) che esegua automaticamente il deployment della scansione di Knowledge Catalog ogni volta che il file di regole viene aggiornato. - Esplora le regole SQL personalizzate:vai oltre i tipi di regole standard. Knowledge Catalog supporta regole SQL personalizzate per applicare una logica più complessa e specifica per l'attività che non può essere espressa con controlli predefiniti. Questa funzionalità consente di personalizzare la convalida in base ai tuoi requisiti unici.
- Ottimizza le scansioni per efficienza e costi: per le tabelle molto grandi, puoi migliorare le prestazioni e ridurre i costi evitando di eseguire sempre la scansione dell'intero set di dati. Esplora l'utilizzo dei filtri per restringere la scansione a intervalli di tempo o segmenti di dati specifici oppure configura le scansioni campionate per controllare una percentuale rappresentativa dei tuoi dati.
- Visualizza i risultati:l'output di ogni scansione della qualità dei dati di Knowledge Catalog viene scritto in una tabella BigQuery. Collega questa tabella a Looker Studio per creare dashboard che monitorano i punteggi di qualità dei dati nel tempo, aggregati in base alle dimensioni che hai definito (ad es. completezza, validità). In questo modo, il monitoraggio è proattivo e visibile a tutti gli stakeholder.
- Condividi le best practice: incoraggia la condivisione delle conoscenze all'interno della tua organizzazione per sfruttare l'esperienza collettiva e migliorare la strategia di qualità dei dati. Promuovere una cultura di fiducia nei dati è fondamentale per sfruttare al meglio gli sforzi di governance.
- Leggi la documentazione: