1. Introduction
Cet atelier de programmation fournit un plan technique aux spécialistes des données. Il décrit une approche "code-first" de la gouvernance des données, montrant comment intégrer une gestion robuste de la qualité et des métadonnées directement dans le cycle de vie du développement. Knowledge Catalog est une data fabric intelligente qui permet aux entreprises de gérer, de surveiller et de gouverner de manière centralisée les données de l'ensemble de leur patrimoine, des lacs de données aux entrepôts.
Cet atelier de programmation montre comment utiliser Knowledge Catalog, BigQuery et l'interface de ligne de commande Antigravity pour aplatir des données complexes, les profiler de manière programmatique, générer des suggestions de règles de qualité des données intelligentes et déployer des analyses de qualité automatisées. L'objectif principal est de dépasser les processus manuels pilotés par l'UI, qui sont sujets aux erreurs et difficiles à mettre à l'échelle, et d'établir à la place un framework "policy-as-code" robuste et contrôlable par version.
Points abordés
- Découvrez comment aplatir les données BigQuery imbriquées à l'aide de vues matérialisées pour permettre un profilage complet.
- Découvrez comment déclencher et gérer par programmation les analyses de profilage Knowledge Catalog à l'aide de la bibliothèque cliente Python Knowledge Catalog.
- Comment exporter les données de profil et les structurer en tant qu'entrée pour un modèle d'IA générative.
- Découvrez comment concevoir une requête pour que l'interface de ligne de commande Antigravity analyse les données de profil et génère un fichier de règles YAML conforme à Knowledge Catalog.
- Importance d'un processus interactif avec intervention humaine (human-in-the-loop, HITL) pour valider les configurations générées par l'IA.
- Déployer les règles générées en tant qu'analyse automatisée de la qualité des données
Prérequis
- Un compte Google Cloud et un projet Google Cloud
- Un navigateur Web tel que Chrome
Concepts clés : les piliers de la qualité des données Knowledge Catalog
Il est essentiel de comprendre les composants de base de Knowledge Catalog pour élaborer une stratégie efficace de qualité des données.
- Analyse de profil de données : job Knowledge Catalog qui analyse les données et génère des métadonnées statistiques, y compris les pourcentages de valeurs nulles, le nombre de valeurs distinctes et les distributions de valeurs. Il s'agit de notre phase de "découverte" programmatique.
- Règles de qualité des données : instructions déclaratives définissant les conditions que vos données doivent respecter (par exemple,
NonNullExpectation,SetExpectation,RangeExpectation). - IA générative pour la suggestion de règles : utilisation d'un grand modèle de langage (comme Gemini) pour analyser un profil de données et suggérer des règles de qualité des données pertinentes. Cela accélère le processus de définition d'un framework de qualité de référence.
- Analyse de la qualité des données : job Knowledge Catalog qui valide les données par rapport à un ensemble de règles prédéfinies ou personnalisées.
- Gouvernance programmatique : thème central de la gestion des contrôles de gouvernance (comme les règles de qualité) en tant que code (par exemple, dans des fichiers YAML et des scripts Python). Cela permet l'automatisation, la gestion des versions et l'intégration dans les pipelines CI/CD.
- Human-in-the-loop (HITL) : point de contrôle essentiel pour intégrer l'expertise et la supervision humaines dans un workflow automatisé. Pour les configurations générées par l'IA, l'intervention humaine est essentielle pour valider l'exactitude, la pertinence commerciale et la sécurité des suggestions avant le déploiement.
2. Préparation
Démarrer Cloud Shell
Bien que Google Cloud puisse être utilisé à distance depuis votre ordinateur portable, nous allons nous servir de Google Cloud Shell pour cet atelier de programmation, un environnement de ligne de commande exécuté dans le cloud.
Dans la console Google Cloud, cliquez sur l'icône Cloud Shell dans la barre d'outils supérieure :
Le provisionnement et la connexion à l'environnement prennent quelques instants seulement. Une fois l'opération terminée, le résultat devrait ressembler à ceci :
Cette machine virtuelle contient tous les outils de développement nécessaires. Elle comprend un répertoire d'accueil persistant de 5 Go et s'exécute sur Google Cloud, ce qui améliore nettement les performances du réseau et l'authentification. Vous pouvez effectuer toutes les tâches de cet atelier de programmation dans un navigateur. Vous n'avez rien à installer.
Activer les API requises et configurer l'environnement
Dans Cloud Shell, assurez-vous que l'ID de votre projet est configuré :
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
Dans l'exemple que nous utilisons, l'emplacement est us (multirégional), car les exemples de données publiques que nous allons utiliser se trouvent également dans la région us (multirégionale). BigQuery exige que les données sources et la table de destination d'une requête se trouvent au même emplacement.
Créer un ensemble de données BigQuery dédié
Créez un ensemble de données BigQuery pour héberger nos exemples de données et nos résultats.
bq --location=us mk --dataset $PROJECT_ID:$DATASET_ID
Préparer les exemples de données
Pour cet atelier de programmation, vous allez utiliser un ensemble de données public contenant des données d'e-commerce obscurcies provenant du Google Merchandise Store. Étant donné que les ensembles de données publics sont en lecture seule, vous devez créer une copie modifiable dans votre propre ensemble de données. La commande bq suivante crée une table ga4_transactions dans votre ensemble de données kc_dq_codelab. Il copie les données d'une seule journée (2021-01-31) pour s'assurer que les analyses s'exécutent rapidement.
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`'
Configurer le répertoire de démonstration
Pour commencer, clonez un dépôt GitHub qui contient la structure de dossier et les fichiers nécessaires pour cet atelier de programmation.
# 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
Ce répertoire est désormais votre zone de travail active. Tous les fichiers suivants seront créés ici.
3. Découverte automatisée des données avec le profilage Knowledge Catalog
Le profilage des données de Knowledge Catalog est un outil puissant qui permet de découvrir automatiquement des informations statistiques sur vos données, telles que les pourcentages de valeurs nulles, l'unicité et les distributions de valeurs. Ce processus est essentiel pour comprendre la structure et la qualité de vos données. Toutefois, une limitation connue du profilage Knowledge Catalog est son incapacité à inspecter entièrement les champs imbriqués ou répétés (par exemple, les types RECORD ou ARRAY) dans une table. Il peut identifier qu'une colonne est un type complexe, mais ne peut pas profiler les champs individuels de cette structure imbriquée.
Pour surmonter ce problème, nous allons aplatir les données dans des vues matérialisées spécialement conçues. Cette stratégie transforme chaque champ en colonne de premier niveau, ce qui permet au Knowledge Catalog de profiler chacun d'eux individuellement.
Comprendre le schéma imbriqué
Commençons par examiner le schéma de notre table source. L'ensemble de données Google Analytics 4 (GA4) contient plusieurs colonnes imbriquées et répétées. Pour récupérer le schéma complet de manière programmatique, y compris toutes les structures imbriquées, vous pouvez utiliser la commande bq show et enregistrer le résultat dans un fichier JSON.
bq show --schema --format=json $PROJECT_ID:$DATASET_ID.$TABLE_ID > bq_schema.json
L'inspection du fichier bq_schema.json révèle des structures complexes telles que des éléments d'appareil, de zone géographique, d'e-commerce et d'enregistrement répétés. Il s'agit des structures qui doivent être aplaties pour un profilage efficace.
Aplatir les données avec des vues matérialisées
La création de vues matérialisées (VM) est la solution la plus efficace et la plus pratique pour relever ce défi lié aux données imbriquées. En précalculant les résultats aplatis, les vues matérialisées offrent des avantages considérables en termes de performances et de coûts des requêtes, tout en fournissant une structure plus simple, de type relationnel, pour les analystes et les outils de profilage.
La première idée qui vient à l'esprit est peut-être d'aplatir le tout en une seule vue géante. Toutefois, cette approche intuitive cache un piège dangereux qui peut entraîner une grave corruption des données. Voyons pourquoi il s'agit d'une erreur critique.
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;
Exécutez maintenant ces modèles à l'aide de l'outil de ligne de commande bq. La commande envsubst lira chaque fichier, remplacera les variables telles que $PROJECT_ID et $DATASET_ID par leurs valeurs issues de votre environnement shell, puis transmettra le code SQL final et valide à la commande 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
Exécuter les analyses de profil via le client Python
Maintenant que nous avons nos vues profilables et aplaties, nous pouvons créer et exécuter des analyses de profilage des données Knowledge Catalog pour chacune d'elles de manière programmatique. Le script Python suivant utilise la bibliothèque cliente google-cloud-dataplex pour automatiser ce processus.
Avant d'exécuter le script, il est essentiel de créer un environnement Python isolé dans le répertoire de votre projet. Cela garantit que les dépendances du projet sont gérées séparément, ce qui évite les conflits avec d'autres packages dans votre environnement Cloud Shell.
# Create the virtual environment
python3 -m venv dq_venv
# Activate the environment
source dq_venv/bin/activate
Installez maintenant la bibliothèque cliente Knowledge Catalog dans l'environnement nouvellement activé.
# Install the Knowledge Catalog client library
pip install google-cloud-dataplex
Une fois l'environnement configuré et la bibliothèque installée, vous êtes prêt à créer le script d'orchestration.
Dans la barre d'outils Cloud Shell, cliquez sur "Ouvrir l'éditeur". Créez un fichier nommé 1_run_scan.py et collez-y le code Python suivant. Si vous clonez le dépôt GitHub, ce fichier se trouve déjà dans votre dossier.
Ce script crée une analyse pour chaque vue matérialisée (si elle n'existe pas déjà), l'exécute, puis interroge jusqu'à ce que tous les jobs d'analyse soient terminés.
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()
Exécutez maintenant le script à partir de votre terminal Cloud Shell.
python 1_run_scan.py
Le script va maintenant orchestrer le profilage de vos trois vues matérialisées et vous fournir des informations sur l'état en temps réel. Une fois cette étape terminée, vous disposerez d'un profil statistique riche et lisible par machine pour chaque vue. Vous serez alors prêt pour la prochaine étape de notre workflow : la génération de règles de qualité des données optimisées par l'IA.
Vous pouvez consulter les analyses de profil terminées dans la console Google Cloud.
- Dans le menu de navigation, accédez à Knowledge Catalog, puis à Qualité et profilage des données dans la section Gérer.
- Vous devriez voir les trois analyses de profil listées, ainsi que l'état actuel de leur job. Vous pouvez cliquer sur une analyse pour afficher les résultats détaillés.
Du profil BigQuery à l'entrée prête pour l'IA
Les analyses de profil Knowledge Catalog ont bien été exécutées. Bien que les résultats soient disponibles dans l'API Knowledge Catalog, pour les utiliser comme entrée pour un modèle d'IA générative, nous devons les extraire dans un fichier local structuré.
Le script Python suivant, 2_dq_profile_save.py, recherche de manière programmatique le dernier job d'analyse de profil réussi pour notre vue mv_ga4_user_session_flat. Il récupère ensuite le résultat complet et détaillé du profil, et l'enregistre en tant que fichier JSON local nommé dq_profile_results.json. Ce fichier servira d'entrée directe pour notre analyse de l'IA à l'étape suivante.
Dans l'éditeur Cloud Shell, créez un fichier nommé 2_dq_profile_save.py et collez-y le code suivant. Comme à l'étape précédente, vous pouvez ignorer la création d'un fichier si vous avez cloné le dépôt.
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()
Exécutez maintenant le script depuis votre terminal :
python 2_dq_profile_save.py
Une fois ces opérations effectuées, votre répertoire doit comporter un nouveau fichier nommé dq_profile_results.json. Ce fichier contient les métadonnées statistiques riches et détaillées que nous utiliserons pour générer des règles de qualité des données. Si vous souhaitez vérifier le contenu de dq_profile_results.json, exécutez la commande suivante :
cat dq_profile_results.json
4. Générer des règles de qualité des données avec la CLI Antigravity
Vous pouvez maintenant utiliser la CLI Antigravity (agy) pour lire les résultats de l'analyse du profil local. Écrire manuellement des spécifications de qualité des données pour des ensembles de données complexes est chronophage et source d'erreurs. L'utilisation d'un agent d'IA générative dans votre terminal accélère ce workflow en rédigeant une configuration déclarative initiale en quelques secondes. Cela permet aux équipes de données de passer de la rédaction manuelle de la syntaxe à une supervision de haut niveau, alignée sur l'activité, avec intervention humaine (HITL).
Pour démarrer l'interface de ligne de commande Antigravity, exécutez la commande suivante dans votre terminal :
agy
Vous êtes maintenant prêt à générer des règles de qualité. Comme l'interface CLI peut lire les fichiers de votre répertoire actuel, elle peut utiliser directement les nouvelles données d'analyse de votre profil.
Demandez à l'agent de créer un plan
Nous allons d'abord demander à l'agent Antigravity d'analyser le profil statistique et de proposer un plan d'action. Nous lui demandons explicitement de ne pas écrire le fichier YAML pour le moment. Il se concentre sur l'analyse et la justification.
Dans votre session interactive Antigravity CLI, saisissez le prompt structuré suivant :
# 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'agent analysera le fichier JSON et renverra un plan structuré semblable à celui-ci :
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']
... │ ... │ ... │ ...
Générer les règles de qualité des données
Il s'agit de l'étape la plus critique de l'ensemble du workflow : l'examen avec intervention humaine (human-in-the-loop, HITL). Le plan généré par l'agent est basé uniquement sur des schémas statistiques dans les données. Il ne comprend pas le contexte de votre activité, les futures modifications de données ni l'intention spécifique derrière vos données. Votre rôle en tant qu'expert humain est de valider, corriger et approuver ce plan avant de le transformer en code.
Examinez attentivement le plan fourni par l'agent.
- Est-ce clair ?
- Est-il conforme à vos connaissances de l'entreprise ?
- Existe-t-il des règles statistiquement valables, mais pratiquement inutiles ?
Les résultats que vous recevez de l'agent peuvent varier. Votre objectif est de l'affiner. Par exemple, imaginons que le plan suggère une règle rowCount, car le tableau comporte un nombre fixe de lignes dans les données d'échantillon. En tant qu'expert humain, vous savez peut-être que la taille de cette table devrait augmenter chaque jour, ce qui rend une règle stricte sur le nombre de lignes peu pratique et susceptible de générer de fausses alertes. C'est un exemple parfait d'application du contexte métier qui manque à l'IA.
Vous allez maintenant fournir des commentaires à l'agent et lui donner la commande finale pour générer le code. Vous devez adapter la requête suivante en fonction du forfait que vous avez réellement reçu et des corrections que vous souhaitez apporter.
La requête ci-dessous est un modèle. La première ligne vous permet de fournir vos corrections spécifiques. Si le plan que l'agent vous a fourni est parfait et n'a pas besoin d'être modifié, vous pouvez simplement supprimer cette ligne.
Dans la même session Antigravity, saisissez votre version adaptée de la requête suivante :
# 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'agent génère alors le contenu YAML en fonction de vos instructions précises et validées par un humain. Une fois l'opération terminée, vous trouverez un nouveau fichier nommé dq_rules.yaml dans votre répertoire de travail.
Créer et exécuter l'analyse de la qualité des données
Maintenant que vous disposez d'un fichier dq_rules.yaml généré par l'agent et validé par un humain, vous pouvez le déployer en toute confiance.
Quittez l'interface de ligne de commande Antigravity en saisissant /quit ou en appuyant deux fois sur Ctrl+C.
La commande gcloud suivante crée une ressource d'analyse de données Knowledge Catalog. Il n'exécute pas encore l'analyse. Il enregistre simplement la définition et la configuration de l'analyse (notre fichier YAML) dans Knowledge Catalog.
Exécutez la commande suivante dans votre terminal :
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"
Maintenant que l'analyse est définie, vous pouvez déclencher un job pour l'exécuter.
gcloud dataplex datascans run $DQ_SCAN --location=$LOCATION --project=$PROJECT_ID
Cette commande génère un ID de job. Vous pouvez surveiller l'état de ce job dans la section Knowledge Catalog de la console Google Cloud. Une fois l'opération terminée, les résultats seront écrits dans une table BigQuery pour analyse.
5. Rôle essentiel de l'approche human-in-the-loop (HITL)
Bien que l'utilisation de l'agent Antigravity pour accélérer la génération de règles soit extrêmement efficace, il est essentiel de considérer l'IA comme un copilote très compétent, et non comme un pilote entièrement autonome. Le processus HITL (Human-in-the-Loop) n'est pas une suggestion facultative. Il s'agit d'une étape fondamentale et non négociable dans tout workflow de gouvernance des données. Le simple déploiement d'artefacts générés par l'IA sans une supervision humaine rigoureuse est voué à l'échec.
Considérez le dq_rules.yaml généré par IA comme une demande d'extraction envoyée par un développeur IA extrêmement rapide, mais inexpérimenté. Il nécessite un examen approfondi de la part d'un expert humain senior (vous) avant de pouvoir être fusionné dans la "branche principale" de votre règlement de gouvernance et déployé. Cet examen est essentiel pour atténuer les faiblesses inhérentes aux grands modèles de langage.
Voici une explication détaillée de la raison pour laquelle cet examen manuel est indispensable et de ce que vous devez rechercher plus précisément :
1. Validation contextuelle : l'IA manque de connaissances sur l'entreprise
- La faiblesse du LLM : un LLM est un expert en modèles et en statistiques, mais il ne comprend absolument pas le contexte de votre entreprise. Par exemple, si une colonne
new_campaign_idprésente un taux de valeurs nulles de 98 %, un LLM peut ignorer cette colonne pour une raison statistique. - Le rôle essentiel de l'humain : vous, l'expert humain, savez que le champ
new_campaign_idvient d'être ajouté hier pour un lancement de produit majeur la semaine prochaine. Vous savez que son ratio de valeur nulle devrait être élevé maintenant, mais qu'il devrait chuter de manière significative. Vous savez également qu'une fois renseignée, elle doit respecter un format spécifique. L'IA ne peut pas inférer ces connaissances externes sur l'entreprise. Votre rôle consiste à appliquer ce contexte commercial aux suggestions statistiques de l'IA, en les remplaçant ou en les complétant si nécessaire.
2. Exactitude et précision : protection contre les hallucinations et les erreurs subtiles
- La faiblesse des LLM : les LLM peuvent être "confiants dans leur erreur". Ils peuvent "halluciner" ou générer du code qui est subtilement incorrect. Par exemple, il peut générer un fichier YAML avec une règle correctement nommée, mais un paramètre non valide, ou il peut mal orthographier un type de règle (par exemple,
setExpectationsau lieu desetExpectation). Ces erreurs subtiles entraîneront l'échec du déploiement, mais peuvent être difficiles à repérer. - Rôle essentiel de l'humain : votre rôle consiste à agir en tant que linter et validateur de schéma ultime. Vous devez vérifier méticuleusement le fichier YAML généré par rapport à la spécification
DataQualityRuleofficielle de Knowledge Catalog. Vous ne vous contentez pas de vérifier si le code "semble correct". Vous validez sa correction syntaxique et sémantique pour vous assurer qu'il est 100% conforme à l'API cible. C'est pourquoi l'atelier de programmation invite l'agent à référencer les fichiers de schéma pour réduire le risque d'erreur, mais la vérification finale vous incombe.
3. Sécurité et réduction des risques : éviter les conséquences en aval
- Point faible du LLM : une règle de qualité des données défectueuse déployée en production peut avoir de graves conséquences. Si l'IA suggère un
rangeExpectationtrop large pour un montant de transaction financière, elle risque de ne pas détecter d'activité frauduleuse. À l'inverse, si elle suggère une règle trop stricte basée sur un petit échantillon de données, elle pourrait inonder votre équipe d'astreinte de milliers d'alertes de faux positifs, ce qui entraînerait une fatigue liée aux alertes et le risque de passer à côté de vrais problèmes. - Le rôle essentiel de l'humain : vous êtes l'ingénieur en sécurité. Vous devez évaluer l'impact potentiel en aval de chaque règle suggérée par l'IA. Posez-vous la question suivante : "Que se passe-t-il si cette règle échoue ? L'alerte est-elle exploitable ? Quel est le risque si cette règle est validée à tort ?" Cette évaluation des risques est une capacité propre à l'humain, qui met en balance le coût de l'échec et l'avantage de la vérification.
4. Gouvernance en tant que processus continu : intégrer des connaissances prospectives
- Faiblesse du LLM : les connaissances de l'IA sont basées sur un instantané statique des données. Le profil est donc issu d'un moment précis. Il n'a aucune connaissance des événements futurs.
- Le rôle essentiel de l'humain : votre stratégie de gouvernance doit être tournée vers l'avenir. Vous savez qu'une source de données doit être migrée le mois prochain, ce qui modifiera le stream_id. Vous savez qu'un nouveau pays est ajouté à la liste
geo_country. Le processus HITL consiste à injecter ces connaissances sur l'état futur, en mettant à jour ou en désactivant temporairement les règles pour éviter les problèmes lors des évolutions techniques ou commerciales prévues. La qualité des données n'est pas une configuration ponctuelle. Il s'agit d'un processus évolutif qui doit être guidé par un humain.
En résumé, HITL est le mécanisme essentiel d'assurance qualité et de sécurité qui transforme la gouvernance basée sur l'IA, qui était une idée nouvelle mais risquée, en une pratique responsable, évolutive et de niveau professionnel. Cela garantit que les règles finales déployées ne sont pas seulement accélérées par l'IA, mais qu'elles sont également validées par des humains, combinant la rapidité des machines à la sagesse et au contexte des experts humains.
Toutefois, cette importance accordée à la supervision humaine ne diminue pas la valeur de l'IA. Au contraire, l'IA générative joue un rôle crucial dans l'accélération du processus HITL lui-même.
Sans IA, un ingénieur de données devrait :
- Écrivez manuellement des requêtes SQL complexes pour profiler les données (par exemple,
COUNT DISTINCT,AVG,MIN,MAXpour chaque colonne). - Analysez minutieusement les résultats feuille de calcul par feuille de calcul.
- Écrire chaque ligne du fichier de règles YAML à partir de zéro, une tâche fastidieuse et sujette aux erreurs
L'IA automatise ces étapes laborieuses et chronophages. Il agit comme un analyste infatigable qui traite instantanément le profil statistique et fournit un "premier jet" bien structuré et complet à 80% de la règle.
Cela change fondamentalement la nature du travail humain. Au lieu de passer des heures à traiter manuellement les données et à coder des modèles, l'expert humain peut se concentrer immédiatement sur les tâches à plus forte valeur ajoutée :
- Appliquer un contexte métier critique.
- Valider l'exactitude de la logique de l'IA.
- Prendre des décisions stratégiques sur les règles qui sont vraiment importantes.
Dans ce partenariat, l'IA gère le "quoi" (quels sont les schémas statistiques ?), ce qui permet à l'humain de se concentrer sur le "pourquoi" (pourquoi ce schéma est-il important pour notre activité ?) et le "et alors" (quelle devrait être notre stratégie ?). L'IA ne remplace donc pas la boucle, mais rend chaque cycle plus rapide, plus intelligent et plus efficace.
6. Nettoyer votre environnement
Pour éviter que les ressources utilisées dans cet atelier de programmation ne soient facturées sur votre compte Google Cloud, supprimez le projet contenant les ressources. Toutefois, si vous souhaitez conserver le projet, vous pouvez supprimer les ressources individuelles que vous avez créées.
Supprimer les analyses Knowledge Catalog
Commencez par supprimer les analyses de profilage et de qualité que vous avez créées. Pour éviter la suppression accidentelle de ressources importantes, ces commandes utilisent les noms spécifiques des analyses créées dans cet atelier de programmation.
# 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
Supprimer l'ensemble de données BigQuery
Supprimez ensuite l'ensemble de données BigQuery. Cette commande est irréversible et utilise l'indicateur -f (force) pour supprimer l'ensemble de données et toutes ses tables sans confirmation.
# Manually type this command to confirm you are deleting the correct dataset
bq rm -r -f --dataset $PROJECT_ID:kc_dq_codelab
7. Félicitations !
Vous avez terminé cet atelier de programmation.
Vous avez créé un workflow de gouvernance des données programmatique de bout en bout. Vous avez commencé par utiliser des vues matérialisées pour aplatir des données BigQuery complexes et les rendre ainsi adaptées à l'analyse. Vous avez ensuite exécuté des analyses de profilage Knowledge Catalog de manière programmatique pour générer des métadonnées statistiques. Plus important encore, vous avez utilisé l'interface de ligne de commande Antigravity pour analyser la sortie du profil et générer intelligemment un artefact "policy-as-code" (dq_rules.yaml). Vous avez ensuite utilisé l'interface de ligne de commande pour déployer cette configuration en tant qu'analyse automatisée de la qualité des données, bouclant ainsi la boucle d'une stratégie de gouvernance moderne et évolutive.
Vous disposez désormais du modèle fondamental pour créer des systèmes de qualité des données fiables, accélérés par l'IA et validés par des humains sur Google Cloud.
Étape suivante
- Intégrer à CI/CD : prenez le fichier
dq_rules.yamlet validez-le dans un dépôt Git. Créez un pipeline CI/CD (par exemple, à l'aide de Cloud Build ou GitHub Actions) qui déploie automatiquement l'analyse Knowledge Catalog chaque fois que le fichier de règles est mis à jour. - Explorer les règles SQL personnalisées : allez au-delà des types de règles standards. Knowledge Catalog est compatible avec les règles SQL personnalisées pour appliquer une logique plus complexe et spécifique à l'entreprise, qui ne peut pas être exprimée avec des vérifications prédéfinies. Cette fonctionnalité permet d'adapter la validation à vos besoins spécifiques.
- Optimisez les analyses pour l'efficacité et les coûts : pour les très grandes tables, vous pouvez améliorer les performances et réduire les coûts en n'analysant pas toujours l'ensemble de données complet. Explorez les filtres pour limiter l'analyse à des périodes ou des segments de données spécifiques, ou configurez des analyses échantillonnées pour vérifier un pourcentage représentatif de vos données.
- Visualisez les résultats : les résultats de chaque analyse de la qualité des données Knowledge Catalog sont écrits dans une table BigQuery. Connectez cette table à Looker Studio pour créer des tableaux de bord qui suivent vos scores de qualité des données au fil du temps, agrégés selon les dimensions que vous avez définies (par exemple, "Exhaustivité" et "Validité"). La surveillance devient ainsi proactive et visible pour toutes les parties prenantes.
- Partagez les bonnes pratiques : encouragez le partage des connaissances au sein de votre organisation pour tirer parti de l'expérience collective et améliorer votre stratégie de qualité des données. Pour exploiter pleinement vos efforts de gouvernance, il est essentiel de favoriser une culture de confiance dans les données.
- Consultez la documentation :