1. Introducción
En este codelab, se proporciona un plan técnico para los profesionales de datos. Describe un enfoque de "código primero" para la administración de datos, y demuestra cómo incorporar una administración sólida de la calidad y los metadatos directamente en el ciclo de vida del desarrollo. En esencia, Knowledge Catalog actúa como un tejido de datos inteligente que permite a las organizaciones administrar, supervisar y controlar de forma centralizada los datos en todo su patrimonio, desde los data lakes hasta los almacenes.
En este codelab, se muestra cómo aprovechar Knowledge Catalog, BigQuery y la CLI de Antigravity para aplanar datos complejos, generar perfiles de forma programática, generar sugerencias inteligentes de reglas de calidad de los datos y, luego, implementar análisis de calidad automatizados. El objetivo principal es ir más allá de los procesos manuales basados en la IU, que son propensos a errores y difíciles de escalar, y, en cambio, establecer un marco de trabajo sólido de "política como código" con control de versiones.
Qué aprenderás
- Cómo aplanar los datos anidados de BigQuery con vistas materializadas para habilitar la creación de perfiles integrales
- Cómo activar y administrar de forma programática los análisis de perfil de Knowledge Catalog con la biblioteca cliente de Python de Knowledge Catalog
- Cómo exportar datos de perfil y estructurarlos como entrada para un modelo de IA generativa
- Cómo diseñar una instrucción para que la CLI de Antigravity analice los datos de perfil y genere un archivo de reglas YAML compatible con Knowledge Catalog
- La importancia de un proceso interactivo con interacción humana (HITL) para validar las configuraciones generadas por IA
- Cómo implementar las reglas generadas como un análisis de calidad de los datos automatizado
Requisitos
- Una cuenta de Google Cloud y un proyecto de Google Cloud
- Un navegador web, como Chrome
Conceptos clave: los pilares de la calidad de los datos de Knowledge Catalog
Comprender los componentes principales de Knowledge Catalog es fundamental para crear una estrategia de calidad de los datos eficaz.
- Análisis de perfil de datos: Es un trabajo de Knowledge Catalog que analiza los datos y genera metadatos estadísticos, incluidos los porcentajes de valores nulos, los recuentos de valores distintos y las distribuciones de valores. Esta es nuestra fase de "descubrimiento" programática.
- Reglas de calidad de los datos: Son declaraciones declarativas que definen las condiciones que deben cumplir tus datos (p.ej.,
NonNullExpectation,SetExpectation,RangeExpectation). - IA generativa para la sugerencia de reglas: Se usa un modelo de lenguaje grande (como Gemini) para analizar un perfil de datos y sugerir reglas de calidad de los datos pertinentes. Esto acelera el proceso de definición de un marco de calidad de referencia.
- Análisis de calidad de los datos: Es un trabajo de Knowledge Catalog que valida los datos según un conjunto de reglas personalizadas o predefinidas.
- Administración programática: Es el tema central de la administración de los controles de administración (como las reglas de calidad) como código (p.ej., en archivos YAML y secuencias de comandos de Python). Esto permite la automatización, el control de versiones y la integración en canalizaciones de CI/CD.
- Interacción humana (HITL): Es el punto de control crítico para integrar la experiencia y la supervisión humanas en un flujo de trabajo automatizado. En el caso de las configuraciones generadas por IA, la HITL es fundamental para validar la corrección, la relevancia comercial y la seguridad de las sugerencias antes de la implementación.
2. Configuración y requisitos
Inicie Cloud Shell
Si bien Google Cloud y Spanner se pueden operar de manera remota desde tu laptop, en este codelab usarás Google Cloud Shell, un entorno de línea de comandos que se ejecuta en la nube.
En Google Cloud Console, haz clic en el ícono de Cloud Shell en la barra de herramientas en la parte superior derecha:
El aprovisionamiento y la conexión al entorno deberían tomar solo unos minutos. Cuando termine el proceso, debería ver algo como lo siguiente:
Esta máquina virtual está cargada con todas las herramientas de desarrollo que necesitarás. Ofrece un directorio principal persistente de 5 GB y se ejecuta en Google Cloud, lo que permite mejorar considerablemente el rendimiento de la red y la autenticación. Todo tu trabajo en este codelab se puede hacer en un navegador. No es necesario que instales nada.
Habilita las APIs obligatorias y configura el entorno
En Cloud Shell, asegúrate de que tu ID del proyecto esté configurado:
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
En el ejemplo, usamos us (multirregión) como la ubicación, ya que los datos de muestra públicos que usaremos también se encuentran en us (multirregión). BigQuery requiere que los datos de origen y la tabla de destino de una consulta residan en la misma ubicación.
Crea un conjunto de datos de BigQuery exclusivo
Crea un nuevo conjunto de datos de BigQuery para alojar nuestros datos de muestra y resultados.
bq --location=us mk --dataset $PROJECT_ID:$DATASET_ID
Prepara los datos de muestra
En este codelab, usarás un conjunto de datos públicos que contiene datos ofuscados de comercio electrónico de Google Merchandise Store. Dado que los conjuntos de datos públicos son de solo lectura, debes crear una copia mutable en tu propio conjunto de datos. El siguiente comando de bq crea una tabla nueva, ga4_transactions, en tu conjunto de datos kc_dq_codelab. Copia datos de un solo día (2021-01-31) para garantizar que los análisis se ejecuten rápidamente.
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`'
Configura el directorio de demostración
Para comenzar, clonarás un repositorio de GitHub que contiene la estructura de carpetas y los archivos de asistencia necesarios para este 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
Este directorio ahora es tu área de trabajo activa. Todos los archivos posteriores se crearán aquí.
3. Descubrimiento de datos automatizado con la creación de perfiles de Knowledge Catalog
La generación de perfiles de datos de Knowledge Catalog es una herramienta eficaz para descubrir automáticamente información estadística sobre tus datos, como porcentajes de valores nulos, singularidad y distribuciones de valores. Este proceso es fundamental para comprender la estructura y la calidad de tus datos. Sin embargo, una limitación conocida de la generación de perfiles de Knowledge Catalog es su incapacidad para inspeccionar por completo los campos anidados o repetidos (p.ej., los tipos RECORD o ARRAY) dentro de una tabla. Puede identificar que una columna es un tipo complejo, pero no puede generar perfiles de los campos individuales dentro de esa estructura anidada.
Para superar este problema, aplanaremos los datos en vistas materializadas creadas específicamente para este fin. Con esta estrategia, cada campo se convierte en una columna de nivel superior, lo que permite que Knowledge Catalog cree un perfil de cada uno de ellos de forma individual.
Información sobre el esquema anidado
Primero, examinemos el esquema de nuestra tabla de origen. El conjunto de datos de Google Analytics 4 (GA4) contiene varias columnas anidadas y repetidas. Para recuperar el esquema completo de forma programática, incluidas todas las estructuras anidadas, puedes usar el comando bq show y guardar el resultado como un archivo JSON.
bq show --schema --format=json $PROJECT_ID:$DATASET_ID.$TABLE_ID > bq_schema.json
La inspección del archivo bq_schema.json revela estructuras complejas, como dispositivo, ubicación geográfica, comercio electrónico y elementos de registro repetidos. Estas son las estructuras que requieren aplanamiento para generar perfiles eficaces.
Cómo aplanar datos con vistas materializadas
Crear vistas materializadas (VM) es la solución más eficaz y práctica para este desafío de datos anidados. Al calcular previamente los resultados aplanados, las MV ofrecen ventajas significativas en el rendimiento y el costo de las consultas, a la vez que proporcionan una estructura más simple y similar a la relacional para los analistas y las herramientas de generación de perfiles.
La primera idea natural podría ser aplanar todo en una sola vista gigante. Sin embargo, este enfoque intuitivo oculta una trampa peligrosa que puede provocar una corrupción grave de los datos. Exploremos por qué este es un error crítico.
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;
Ahora, ejecuta estas plantillas con la herramienta de línea de comandos de bq. El comando envsubst leerá cada archivo, sustituirá las variables como $PROJECT_ID y $DATASET_ID por sus valores del entorno de shell y canalizará el SQL final y válido al comando de consulta bq.
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
Ejecuta los análisis de perfil a través del cliente de Python
Ahora que tenemos nuestras vistas aplanadas y aptas para el perfilado, podemos crear y ejecutar de forma programática análisis del perfil de datos de Knowledge Catalog para cada una de ellas. La siguiente secuencia de comandos de Python usa la biblioteca cliente de google-cloud-dataplex para automatizar este proceso.
Antes de ejecutar la secuencia de comandos, es una práctica recomendada fundamental crear un entorno aislado de Python dentro del directorio de tu proyecto. Esto garantiza que las dependencias del proyecto se administren por separado, lo que evita conflictos con otros paquetes en tu entorno de Cloud Shell.
# Create the virtual environment
python3 -m venv dq_venv
# Activate the environment
source dq_venv/bin/activate
Ahora, instala la biblioteca cliente de Knowledge Catalog en el entorno recién activado.
# Install the Knowledge Catalog client library
pip install google-cloud-dataplex
Con el entorno configurado y la biblioteca instalada, ya puedes crear la secuencia de comandos de orquestación.
En la barra de herramientas de Cloud Shell, haz clic en Abrir editor. Crea un archivo nuevo llamado 1_run_scan.py y pega el siguiente código de Python en él. Si clonas el repositorio de GitHub, este archivo ya estará en tu carpeta.
Esta secuencia de comandos creará un análisis para cada vista materializada (si aún no existe), ejecutará el análisis y, luego, sondeará hasta que se completen todos los trabajos de análisis.
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()
Ahora, ejecuta el script desde tu terminal de Cloud Shell.
python 1_run_scan.py
Ahora, la secuencia de comandos coordinará la generación de perfiles de tus tres vistas materializadas y proporcionará actualizaciones de estado en tiempo real. Una vez que se complete, tendrás un perfil estadístico enriquecido y legible por máquina para cada vista, listo para la siguiente etapa de nuestro flujo de trabajo: la generación de reglas de calidad de los datos potenciadas por IA.
Puedes ver los análisis de perfil completados en la consola de Google Cloud.
- En el menú de navegación, ve a Knowledge Catalog y Calidad y creación de perfiles de datos en la sección Administrar.
- Deberías ver los tres análisis de perfil, junto con el estado del trabajo más reciente. Puedes hacer clic en un análisis para ver los resultados detallados.
Del perfil de BigQuery a la entrada lista para IA
Se ejecutaron correctamente los análisis de perfil de Knowledge Catalog. Si bien los resultados están disponibles en la API de Knowledge Catalog, para usarlos como entrada para un modelo de IA generativa, debemos extraerlos en un archivo local estructurado.
La siguiente secuencia de comandos de Python, 2_dq_profile_save.py, encuentra de forma programática el trabajo de análisis de perfil exitoso más reciente para nuestra vista mv_ga4_user_session_flat. Luego, recupera el resultado del perfil completo y detallado, y lo guarda como un archivo JSON local llamado dq_profile_results.json. Este archivo servirá como entrada directa para nuestro análisis de IA en el siguiente paso.
En el editor de Cloud Shell, crea un archivo nuevo llamado 2_dq_profile_save.py y pega el siguiente código en él. Al igual que en el paso anterior, puedes omitir la creación de un archivo si clonaste el repositorio.
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()
Ahora, ejecuta la secuencia de comandos desde la terminal:
python 2_dq_profile_save.py
Cuando el proceso finalice con éxito, tendrás un archivo nuevo llamado dq_profile_results.json en tu directorio. Este archivo contiene los metadatos estadísticos detallados y enriquecidos que usaremos para generar reglas de calidad de los datos. Si deseas verificar el contenido de dq_profile_results.json, ejecuta el siguiente comando:
cat dq_profile_results.json
4. Genera reglas de calidad de los datos con la CLI de Antigravity
Ahora puedes usar la CLI de Antigravity (agy) para leer los resultados del análisis del perfil local. Escribir manualmente las especificaciones de calidad de los datos para conjuntos de datos complejos lleva mucho tiempo y es propenso a errores. Usar un agente de IA generativa en tu terminal acelera este flujo de trabajo, ya que redacta una configuración declarativa inicial en segundos. Esto permite que los equipos de datos pasen de la redacción manual de sintaxis a la supervisión de alto nivel y alineada con la empresa de la interacción humana (HITL).
Para iniciar la CLI de Antigravity, ejecuta el siguiente comando en tu terminal:
agy
Ahora puedes generar reglas de calidad. Dado que la CLI puede leer archivos en tu directorio actual, puede usar directamente los datos de análisis de tu nuevo perfil.
Dale instrucciones al agente para que cree un plan
Primero, le pediremos al agente de Antigravity que analice el perfil estadístico y proponga un plan de acción. Le indicamos explícitamente que aún no escriba el archivo YAML. Esto centra su atención en el análisis y la justificación.
En tu sesión interactiva de la CLI de Antigravity, ingresa la siguiente instrucción estructurada:
# 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.
El agente analizará el archivo JSON y debería devolver un plan estructurado similar a este:
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 las reglas de calidad de los datos
Este es el paso más importante de todo el flujo de trabajo: la revisión con interacción humana (HITL). El plan que genera el agente se basa únicamente en patrones estadísticos de los datos. No comprende el contexto de tu empresa, los cambios futuros en los datos ni la intención específica detrás de tus datos. Tu función como experto humano es validar, corregir y aprobar este plan antes de convertirlo en código.
Revisa con atención el plan que te proporcionó el agente.
- ¿Tiene sentido?
- ¿Se alinea con tus conocimientos empresariales?
- ¿Existen reglas estadísticamente sólidas, pero prácticamente inútiles?
El resultado que recibas del agente puede variar. Tu objetivo es perfeccionarlo. Por ejemplo, imagina que el plan sugiere una regla de rowCount porque la tabla tiene una cantidad fija de filas en los datos de muestra. Como experto humano, es posible que sepas que se espera que el tamaño de esta tabla aumente a diario, lo que hace que una regla estricta de recuento de filas sea poco práctica y propensa a generar alertas falsas. Este es un ejemplo perfecto de la aplicación del contexto empresarial del que carece la IA.
Ahora, le proporcionarás comentarios al agente y le darás el comando final para generar el código. Debes adaptar la siguiente instrucción según el plan que recibiste y las correcciones que quieras realizar.
La instrucción a continuación es una plantilla. En la primera línea, proporcionarás tus correcciones específicas. Si el plan que te dio el agente es perfecto y no necesita cambios, puedes borrar esa línea.
En la misma sesión de Antigravity, ingresa tu versión adaptada de la siguiente instrucción:
# 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.
Ahora, el agente generará el contenido en formato YAML según tus instrucciones precisas y validadas por humanos. Cuando se complete el proceso, encontrarás un archivo nuevo llamado dq_rules.yaml en tu directorio de trabajo.
Crea y ejecuta el análisis de calidad de los datos
Ahora que tienes un archivo dq_rules.yaml generado por un agente y validado por un humano, puedes implementarlo con confianza.
Para salir de la CLI de Antigravity, escribe /quit o presiona Ctrl+C dos veces.
El siguiente comando de gcloud crea un recurso nuevo de análisis de datos de Knowledge Catalog. Aún no se ejecuta el análisis, sino que simplemente se registra la definición y la configuración del análisis (nuestro archivo YAML) en Knowledge Catalog.
Ejecuta este comando en tu 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"
Ahora que se definió el análisis, puedes activar un trabajo para ejecutarlo.
gcloud dataplex datascans run $DQ_SCAN --location=$LOCATION --project=$PROJECT_ID
Este comando generará un ID de trabajo. Puedes supervisar el estado de este trabajo en la sección Knowledge Catalog de la consola de Google Cloud. Una vez que se complete el proceso, los resultados se escribirán en una tabla de BigQuery para su análisis.
5. El rol fundamental de la interacción humana (HITL)
Si bien usar el agente de Antigravity para acelerar la generación de reglas es increíblemente potente, es fundamental tratar a la IA como un copiloto altamente capacitado, no como un piloto completamente autónomo. El proceso de interacción humana (HITL) no es una sugerencia opcional, sino un paso fundamental e innegociable en cualquier flujo de trabajo de administración de datos. Implementar artefactos generados por IA sin una supervisión humana rigurosa es una receta para el fracaso.
Piensa en el dq_rules.yaml generado por IA como una solicitud de extracción enviada por un desarrollador de IA extremadamente rápido, pero sin experiencia. Requiere una revisión exhaustiva de un experto humano sénior (tú) antes de que se pueda combinar con la "rama principal" de tu política de gobierno y se pueda implementar. Esta revisión es fundamental para mitigar las debilidades inherentes de los modelos de lenguaje grandes.
A continuación, se incluye un desglose detallado de por qué esta revisión humana es indispensable y qué debes buscar específicamente:
1. Validación contextual: La IA no tiene conocimiento de la empresa
- La debilidad del LLM: Un LLM es un experto en patrones y estadísticas, pero no comprende el contexto de tu empresa. Por ejemplo, si una columna,
new_campaign_id, tiene un 98% de valores nulos, un LLM podría ignorarla por un motivo estadístico. - El rol fundamental del ser humano: Tú, el experto humano, sabes que el campo
new_campaign_idse agregó ayer para el lanzamiento de un producto importante la próxima semana. Sabes que su proporción de nulos debería ser alta ahora, pero se espera que disminuya significativamente. También sabes que, una vez que se completa, debe seguir un formato específico. La IA no puede inferir este conocimiento comercial externo. Tu función es aplicar este contexto comercial a las sugerencias estadísticas de la IA, y anularlas o aumentarlas según sea necesario.
2. Corrección y precisión: Protección contra alucinaciones y errores sutiles
- La debilidad de los LLM: Los LLM pueden ser "confiadamente incorrectos". Pueden "alucinar" o generar código que es sutilmente incorrecto. Por ejemplo, podría generar un archivo YAML con una regla con el nombre correcto, pero con un parámetro no válido, o podría escribir mal un tipo de regla (p.ej.,
setExpectationsen lugar delsetExpectationcorrecto). Estos errores sutiles harán que falle la implementación, pero pueden ser difíciles de detectar. - El rol fundamental del humano: Tu trabajo es actuar como el validador de esquemas y el linter definitivo. Debes verificar meticulosamente el archivo YAML generado en comparación con la especificación oficial de Knowledge Catalog
DataQualityRule. No solo verificas si "se ve bien", sino que validas su corrección sintáctica y semántica para asegurarte de que cumpla con la API de destino en un 100%. Por eso, el codelab le solicita al agente que haga referencia a los archivos de esquema para reducir la posibilidad de errores, pero la verificación final depende de ti.
3. Seguridad y mitigación de riesgos: prevención de consecuencias posteriores
- La debilidad del LLM: Una regla de calidad de los datos defectuosa implementada en producción puede tener consecuencias graves. Si la IA sugiere un
rangeExpectationpara un importe de transacción financiera que es demasiado amplio, es posible que no detecte actividad fraudulenta. Por el contrario, si sugiere una regla demasiado estricta basada en una pequeña muestra de datos, podría inundar a tu equipo de guardia con miles de alertas de falsos positivos, lo que provocaría fatiga por alertas y haría que se pasen por alto problemas reales. - El rol fundamental de la persona: Tú eres el ingeniero de seguridad. Debes evaluar el posible impacto posterior de cada regla que sugiera la IA. Pregúntate: "¿Qué sucede si falla esta regla? ¿La alerta es práctica? ¿Cuál es el riesgo si esta regla se aprueba de forma incorrecta?" Esta evaluación de riesgos es una capacidad exclusivamente humana que sopesa el costo del error en comparación con el beneficio de la verificación.
4. Administración como proceso continuo: incorporación de conocimientos prospectivos
- La debilidad del LLM: El conocimiento de la IA se basa en una instantánea estática de los datos: los resultados del perfil provienen de un momento específico. No tiene conocimiento de eventos futuros.
- El rol fundamental de las personas: Tu estrategia de gobernanza debe ser prospectiva. Sabes que una fuente de datos se migrará el próximo mes, lo que cambiará el stream_id. Sabes que se agregará un país nuevo a la lista de
geo_country. El proceso de HITL es donde se inyecta este conocimiento del estado futuro, ya que se actualizan o inhabilitan temporalmente las reglas para evitar interrupciones durante las evoluciones técnicas o comerciales planificadas. La calidad de los datos no es una configuración única, sino un proceso vivo que debe evolucionar, y solo un ser humano puede guiar esa evolución.
En resumen, la HITL es el mecanismo esencial de control de calidad y seguridad que transforma la administración impulsada por IA de una idea novedosa pero riesgosa en una práctica responsable, escalable y de nivel empresarial. Esto garantiza que las políticas finales implementadas no solo se aceleren con la IA, sino que también se validen por humanos, lo que combina la velocidad de las máquinas con la sabiduría y el contexto de los expertos humanos.
Sin embargo, este énfasis en la supervisión humana no disminuye el valor de la IA. Por el contrario, la IA generativa desempeña un papel fundamental en la aceleración del proceso de HITL en sí.
Sin la IA, un ingeniero de datos tendría que hacer lo siguiente:
- Escribe manualmente consultas de SQL complejas para generar perfiles de los datos (p.ej.,
COUNT DISTINCT,AVG,MIN,MAXpara cada columna). - Analizar minuciosamente los resultados hoja de cálculo por hoja de cálculo
- Escribir cada línea del archivo de reglas YAML desde cero, una tarea tediosa y propensa a errores
La IA automatiza estos pasos laboriosos y que consumen mucho tiempo. Actúa como un analista incansable que procesa al instante el perfil estadístico y proporciona un "primer borrador" de la política bien estructurado y completo en un 80%.
Esto cambia fundamentalmente la naturaleza del trabajo humano. En lugar de pasar horas procesando datos de forma manual y escribiendo código estándar, el experto humano puede enfocarse de inmediato en las tareas de mayor valor:
- Aplicar un contexto empresarial crítico
- Validar la corrección de la lógica de la IA
- Tomar decisiones estratégicas sobre qué reglas son realmente importantes
En esta asociación, la IA se encarga del "qué" (¿cuáles son los patrones estadísticos?), lo que libera al humano para que se concentre en el "por qué" (¿por qué este patrón es importante para nuestro negocio?) y el "y entonces" (¿entonces cuál debería ser nuestra política?). Por lo tanto, la IA no reemplaza el ciclo, sino que hace que cada ciclo sea más rápido, inteligente y eficaz.
6. Limpia tu entorno
Para evitar que se apliquen cargos futuros a tu cuenta de Google Cloud por los recursos que usaste en este codelab, debes borrar el proyecto que contiene los recursos. Sin embargo, si deseas conservar el proyecto, puedes borrar los recursos individuales que creaste.
Borra los análisis de Knowledge Catalog
Primero, borra los análisis de perfil y de calidad que creaste. Para evitar el borrado accidental de recursos importantes, estos comandos usan los nombres específicos de los análisis creados en este 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
Borra el conjunto de datos de BigQuery
A continuación, borra el conjunto de datos de BigQuery. Este comando es irreversible y usa la marca -f (forzar) para quitar el conjunto de datos y todas sus tablas sin confirmación.
# Manually type this command to confirm you are deleting the correct dataset
bq rm -r -f --dataset $PROJECT_ID:kc_dq_codelab
7. ¡Felicitaciones!
Completaste el codelab correctamente.
Compilaste un flujo de trabajo de administración de datos programático de extremo a extremo. Comenzaste por usar vistas materializadas para aplanar datos complejos de BigQuery, lo que los hizo adecuados para el análisis. Luego, ejecutaste de forma programática análisis de perfiles de Knowledge Catalog para generar metadatos estadísticos. Lo más importante es que aprovechaste la CLI de Antigravity para analizar el resultado del perfil y generar de forma inteligente un artefacto de "política como código" (dq_rules.yaml). Luego, usaste la CLI para implementar esta configuración como un análisis de calidad de los datos automatizado, lo que cerró el ciclo de una estrategia de administración moderna y escalable.
Ahora tienes el patrón fundamental para crear sistemas de calidad de datos confiables, acelerados por IA y validados por humanos en Google Cloud.
Próximos pasos
- Integración con CI/CD: Toma el archivo
dq_rules.yamly confirma su incorporación a un repositorio de Git. Crea una canalización de CI/CD (p.ej., con Cloud Build o GitHub Actions) que implemente automáticamente el análisis de Knowledge Catalog cada vez que se actualice el archivo de reglas. - Explora las reglas de SQL personalizadas: Ve más allá de los tipos de reglas estándar. Knowledge Catalog admite reglas de SQL personalizadas para aplicar una lógica más compleja y específica de la empresa que no se puede expresar con verificaciones predefinidas. Esta función permite adaptar la validación a tus requisitos únicos.
- Optimiza los análisis para mejorar la eficiencia y reducir los costos: En el caso de las tablas muy grandes, puedes mejorar el rendimiento y reducir los costos si no analizas siempre todo el conjunto de datos. Explora con filtros para limitar el análisis a períodos o segmentos de datos específicos, o bien configura análisis muestreados para verificar un porcentaje representativo de tus datos.
- Visualiza los resultados: El resultado de cada análisis de calidad de los datos de Knowledge Catalog se escribe en una tabla de BigQuery. Conecta esta tabla a Looker Studio para crear paneles que hagan un seguimiento de tus puntuaciones de calidad de los datos a lo largo del tiempo, agregadas según las dimensiones que definiste (p.ej., integridad y validez). Esto hace que la supervisión sea proactiva y visible para todas las partes interesadas.
- Comparte prácticas recomendadas: Fomenta el intercambio de conocimientos dentro de tu organización para aprovechar la experiencia colectiva y mejorar tu estrategia de calidad de los datos. Fomentar una cultura de confianza en los datos es clave para aprovechar al máximo tus esfuerzos de administración.
- Lee la documentación: