Calidad de los datos programática con Dataplex y la IA generativa

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, Dataplex Universal Catalog actúa como un tejido de datos inteligente, lo 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 Dataplex, BigQuery y la CLI de Gemini 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.

Requisitos previos

• Conocimientos básicos sobre la consola de Google Cloud
• Habilidades básicas de la interfaz de línea de comandos y de Google Cloud Shell

Qué aprenderás

• Cómo aplanar los datos anidados de BigQuery con vistas materializadas para habilitar la generación de perfiles integral
• Cómo activar y administrar de forma programática los análisis de perfil de Dataplex con la biblioteca cliente de Python de Dataplex
• 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 Gemini analice los datos de perfil y genere un archivo de reglas YAML compatible con Dataplex
• 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 Dataplex

Comprender los componentes principales de Dataplex es fundamental para crear una estrategia de calidad de los datos eficaz.

• Análisis de perfil de datos: Es un trabajo de Dataplex 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 y 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 referencia de calidad.
• Análisis de calidad de los datos: Es un trabajo de Dataplex que valida los datos en función de 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 supervisión y la experiencia 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

Inicia 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="dataplex_dq_codelab"
export TABLE_ID="ga4_transactions"

gcloud services enable dataplex.googleapis.com \
                       bigquery.googleapis.com \
                       serviceusage.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 dedicado

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úblico 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 dataplex_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.

git clone https://github.com/GoogleCloudPlatform/devrel-demos
cd devrel-demos/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 Dataplex

La generación de perfiles de datos de Dataplex es una herramienta potente 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 Dataplex es su incapacidad para inspeccionar por completo los campos anidados o repetidos (p.ej., Tipos RECORD o ARRAY) dentro de una tabla. Puede identificar que una columna es de tipo complejo, pero no puede generar un perfil 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 Dataplex genere 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

💡 Sugerencia: También puedes explorar este esquema anidado de forma visual en la consola de Google Cloud. Navega a BigQuery, busca tu conjunto de datos dataplex_dq_codelab, selecciona la tabla ga4_transactions y haz clic en la pestaña Esquema.

Si inspeccionas el archivo bq_schema.json, verás estructuras complejas, como elementos de registros repetidos, de dispositivos, de ubicación geográfica y de comercio electrónico. Estas son las estructuras que requieren aplanamiento para una generación de perfiles eficaz.

Cómo aplanar datos con vistas materializadas

Crear vistas materializadas (MV) 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.

💡 La trampa del producto cartesiano: Por qué falla una sola vista aplanada

Si aplanas una tabla desanidando varios arrays a la vez, se crea un producto cartesiano (unión cruzada). Este es un error crítico que no solo ralentiza las consultas, sino que también daña tus datos.

Por ejemplo, un solo evento con 3 elementos y 5 parámetros se expandiría a 15 filas. Esto haría que cualquier métrica a nivel del evento (como los ingresos) se multiplicara por 15 y que los recuentos de eventos fueran inútiles. Se pierde el significado de cada fila.

La estrategia correcta es crear varias vistas específicas, en las que cada vista conserve una granularidad analítica única y clara.

Piensa en esto como una "renormalización" de la tabla de GA4 desnormalizada para diferentes fines de análisis, de manera similar a como tendrías tablas separadas de pedidos y order_items en una base de datos tradicional.

Por lo tanto, para este codelab, implementaremos esta estrategia creando tres vistas materializadas distintas:

• Vista de sesión: Una fila por evento.
• Vista de transacciones: Una fila por transacción
• Vista de elementos: Una fila por elemento.

Este enfoque garantiza la precisión de los datos para la generación de perfiles y el análisis.

1. 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`;

2. 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;

3. 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 de 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 de perfil de datos de Dataplex 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 Dataplex en el entorno recién activado.

# Install the Dataplex 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_dataplex_scans.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.

⚠️ Revisa tu entorno

Las sesiones de Cloud Shell pueden agotar el tiempo de espera. Antes de ejecutar la secuencia de comandos, verifica que las variables de entorno estén configuradas correctamente ejecutando echo $PROJECT_ID. Si devuelve una línea vacía, vuelve a ejecutar los comandos de exportación de la sección "Configuración y requisitos".

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_dataplex_scans.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.

1. En el menú de navegación, ve a Dataplex Universal Catalog y Perfil en la sección Gobernanza.

2. 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 la IA

Los análisis de perfil de Dataplex se ejecutaron correctamente. Si bien los resultados están disponibles en la API de Dataplex, para usarlos como entrada para un modelo de IA generativa, debemos extraerlos en un archivo estructurado local.

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.

⚠️ Revisa tu entorno

Las sesiones de Cloud Shell pueden agotar el tiempo de espera. Antes de ejecutar la secuencia de comandos, verifica que las variables de entorno estén configuradas correctamente ejecutando echo $PROJECT_ID. Si devuelve una línea vacía, vuelve a ejecutar los comandos de exportación de la sección "Configuración y requisitos".

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 Dataplex 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 'run_dataplex_scans.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. Cómo generar reglas de calidad de los datos con la CLI de Gemini

Instala y configura la CLI de Gemini

Si bien puedes llamar a la API de Gemini de forma programática, usar una herramienta como la CLI de Gemini ofrece una forma interactiva y potente de integrar la IA generativa directamente en tus flujos de trabajo de terminal. La CLI de Gemini no es solo un chatbot, sino una herramienta de flujo de trabajo de línea de comandos que puede leer tus archivos locales, comprender tu código y, además, interactuar con otras herramientas del sistema, como gcloud, para automatizar tareas complejas. Esto lo hace ideal para nuestro caso de uso.

Requisitos previos

Primero, asegúrate de cumplir con el requisito previo necesario: Node.js versión 20 o una posterior debe estar instalado en tu entorno de Cloud Shell. Para verificar tu versión, ejecuta node -v.

Instalación

Hay dos formas de usar la CLI de Gemini: instalarla de forma temporal o más permanente. Aquí abordaremos ambos métodos.

Puedes ejecutar la CLI de Gemini directamente para una sola sesión sin necesidad de realizar ninguna instalación permanente. Esta es la forma más limpia y rápida de "probarlo", ya que mantiene tu entorno completamente sin modificar.

En la terminal de Cloud Shell, ejecuta el siguiente comando:

npx https://github.com/google-gemini/gemini-cli

Este comando descarga y ejecuta el paquete de la CLI de forma temporal.

Para cualquier proyecto real, la práctica recomendada es instalar la CLI de forma local en el directorio del proyecto. Este enfoque tiene varias ventajas clave:

• Aislamiento de dependencias: Garantiza que tu proyecto tenga su propia versión de la CLI, lo que evita conflictos de versiones con otros proyectos.
• Reproducibilidad: Cualquier persona que clone tu proyecto puede instalar las mismas dependencias exactas, lo que hace que tu configuración sea confiable y portátil.
• Alineación con las prácticas recomendadas: Sigue el modelo estándar para administrar las dependencias de proyectos de Node.js, lo que evita los inconvenientes de las instalaciones globales (-g).

Para instalar la CLI de forma local, ejecuta el siguiente comando desde la carpeta de tu proyecto (programmatic-dq):

npm install @google/gemini-cli

Esto crea una carpeta node_modules dentro de programmatic-dq. Para ejecutar la versión que acabas de instalar, usa el comando npx.

npx gemini

Configuración por primera vez

Cualquiera sea el método que elijas, la primera vez que inicies la CLI, te guiará por un proceso de configuración único.

Se te pedirá que elijas un tema de color y, luego, que te autentiques. El método más sencillo es acceder con tu Cuenta de Google cuando se te solicite. El nivel gratuito proporcionado es suficiente para este codelab.

Ahora que la CLI está instalada y configurada, puedes continuar con la generación de las reglas. La CLI conoce los archivos de su directorio actual, lo que es fundamental para el siguiente paso.

Genera las reglas de calidad de los datos

Si bien es posible pedirle a un LLM que genere un archivo de configuración de una sola vez, la naturaleza no determinística de los modelos generativos implica que el resultado no siempre cumplirá a la perfección con el esquema estricto que requieren herramientas como gcloud. Un método más confiable es un proceso interactivo de varios pasos en el que, primero, la IA actúa como analista para proponer un plan, tú (el experto humano) revisas y apruebas ese plan, y solo entonces la IA genera el código final según tus instrucciones aprobadas.

Este enfoque transforma el proceso de un simple comando en una sesión colaborativa, lo que garantiza que el resultado final sea preciso, validado y esté listo para la implementación.

Primero, le pediremos a Gemini que analice el perfil estadístico y actúe como un ingeniero de datos experto, proponiendo 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 Gemini, pega la siguiente instrucción:

💡 Sugerencia de instrucción: Observa el uso de ./dq_profile_results.json. Si especificamos la ruta de acceso con ./, le indicamos explícitamente a la CLI que busque el archivo en el directorio de trabajo actual. Esto hace que la instrucción sea más sólida y ayuda a evitar errores si la CLI se inició accidentalmente desde la ubicación incorrecta.

You are an expert Google Cloud Dataplex engineer. 
Your first task is to create a plan. I have a file in the current directory named ./dq_profile_results.json. 
Based on the statistical data within that file, propose a step-by-step plan to create a Dataplex data quality rules file. 
Your plan should identify which specific columns are good candidates for rules like nonNullExpectation, setExpectation, or rangeExpectation, and explain why based on the metrics (e.g., "Plan to create a nonNullExpectation for column X because its null percentage is 0%"). 
Do not write the YAML file yet. Just provide the plan.

Gemini analizará el archivo JSON y debería devolver un plan estructurado similar a este:

Plan to create Dataplex data quality rules

1. Create a row count rule:
  - Analysis: The profile shows a total of 26,489 rows.
  - Proposed Rule: I will implement a rowCount rule to ensure the table's row count remains within a healthy range (e.g., 20,000 to 30,000). This is a fundamental check to guard against major data loading failures or unexpected deletions.
2. Implement non-null expectation rules:
  - Analysis: The profiling results show that many columns have no null values. For example, event_date, event_timestamp, event_name, user_pseudo_id, stream_id, and platform all appear to be fully populated.
  - Proposed Rules: I will create nonNullExpectation rules for columns where the nullRatio is 0 or not present, indicating they are always expected to contain a value:
    - event_date
    - event_timestamp
    - event_name
    - user_pseudo_id
    - stream_id
    - platform
    - device_category
    - device_os
3. Implement set expectation rules:
  - Analysis: Several columns contain a small, finite set of values, making them ideal for set validation.
  - Proposed Rules: I will create setExpectation rules for these columns:
    - platform: It only contains the value "WEB".
    - device_category: It contains only "desktop", "mobile", and "tablet".
    - geo_continent: It has a well-defined list of continents.
traffic_source_medium: It has a small list of valid mediums like "organic", "(none)", and "referral".
4. Implement range expectation rules:
  - Analysis: The profile provides minimum and maximum values for numeric and string length fields.
  - Proposed Rules: I will create rangeExpectation rules for the following:
    - event_timestamp: The profile shows a clear min and max. I will use these values to create a range check.
    - event_date (String Length): The profile indicates that the minLength and maxLength are both 8, which is perfect for enforcing a YYYYMMDD format.

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 Gemini 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 cuidadosamente el plan que te proporcionó Gemini.

• ¿Tiene sentido?
• ¿Se alinea con tus conocimientos empresariales?
• ¿Existen reglas que sean estadísticamente sólidas, pero prácticamente inútiles?

El resultado que recibas de Gemini 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 crezca 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 a Gemini 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.

El mensaje a continuación es una plantilla. En la primera línea, proporcionarás tus correcciones específicas. Si el plan que te dio Gemini es perfecto y no necesita cambios, puedes borrar esa línea.

En la misma sesión de Gemini, pega la versión adaptada de la siguiente instrucción:

[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'."]

Once you have incorporated my feedback, please generate the `dq_rules.yaml` file.

You must adhere to the following strict requirements:

- Schema Compliance: The YAML structure must strictly conform to the DataQualityRule specification. For a definitive source of truth, you must 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.

- Data-Driven Values: All rule parameters, such as thresholds or expected values, must be derived directly from the statistical metrics in dq_profile_results.json.

- Rule Justification: For each rule, add a comment (#) on the line above explaining the justification, as you outlined in your plan.

- Output Purity: The final output must only be the raw YAML code block, perfectly formatted and ready for immediate deployment.

Ahora Gemini generará el contenido en YAML según tus instrucciones precisas y validadas por personas. 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 IA y validado por humanos, puedes implementarlo con confianza.

Para salir de la CLI de Gemini, escribe /quit o presiona Ctrl+C dos veces.

El siguiente comando de gcloud crea un recurso de análisis de datos de Dataplex nuevo. 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 Dataplex.

Ejecuta este comando en tu terminal:

export DQ_SCAN="dq-scan"
gcloud dataplex datascans create data-quality $DQ_SCAN \
    --project=$PROJECT_ID \
    --location=$REGION \
    --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=$REGION --project=$PROJECT_ID

Este comando generará un ID de trabajo. Puedes supervisar el estado de este trabajo en la sección Dataplex de la consola de Google Cloud. Una vez que se complete, 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 Gemini 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 sólido y confiable. 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. Antes de que se pueda combinar con la "rama principal" de tu política de gobernanza y se pueda implementar, requiere una revisión exhaustiva de un experto humano sénior, es decir, tú. 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 del negocio

• 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 humano: Tú, el experto humano, sabes que el campo new_campaign_id se 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., setExpectations en lugar del setExpectation correcto). 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 DataQualityRule de Dataplex. 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 a Gemini 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 rangeExpectation para 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 generarí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 de la falla con el beneficio de la verificación.

4. La gobernanza 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 los humanos: Tu estrategia de administración 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 seguridad y garantía de calidad 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 IA, sino que también se validen por personas, 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:

1. Escribir manualmente consultas de SQL complejas para generar perfiles de los datos (p.ej., COUNT DISTINCT, AVG, MIN, MAX para cada columna).
2. Analizar minuciosamente los resultados hoja de cálculo por hoja de cálculo
3. 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 manualmente y escribiendo código repetitivo, el experto humano puede enfocarse de inmediato en las tareas más valiosas:

• Aplicar el contexto comercial 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 permite que el humano se concentre en el "por qué" (¿por qué este patrón es importante para nuestra empresa?) 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 iteración sea más rápida, 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 Dataplex

Primero, borra los análisis de calidad y de perfil 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 data-scans delete profile-scan-mv-ga4-user-session-flat \
    --location=us-central1 \
    --project=$PROJECT_ID --quiet

gcloud dataplex data-scans delete profile-scan-mv-ga4-ecommerce-transactions \
    --location=us-central1 \
    --project=$PROJECT_ID --quiet

gcloud dataplex data-scans 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.

⚠️ Advertencia: Para evitar la pérdida accidental de datos, te recomendamos que escribas el siguiente comando de forma manual en lugar de copiarlo y pegarlo, y que te asegures de que el nombre del conjunto de datos sea correcto.

# Manually type this command to confirm you are deleting the correct dataset
bq rm -r -f --dataset $PROJECT_ID:dataplex_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 Dataplex para generar metadatos estadísticos. Lo más importante es que aprovechaste la CLI de Gemini 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.yaml y confirma su incorporación en 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 Dataplex 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. Dataplex 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 es una función potente para 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 Dataplex 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 por 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:
  • Documentación oficial de la calidad de los datos de Dataplex
  • CLI de Gemini: Tu agente de IA de código abierto
  • Muestra de código en GitHub