Tutorial Google Cloud Data Agent Kit en VS Code: De Cero a Producción

El Paradigma del Desarrollo Asistido por Agentes en GCP

El desarrollo de canalizaciones analíticas ha enfrentado históricamente una severa fragmentación de contexto: los ingenieros deben alternar continuamente entre el editor de código local (VS Code), la consola web de Google Cloud, clientes SQL para BigQuery y herramientas de orquestación de flujos. La extensión Google Cloud Data Agent Kit cierra esta brecha integrando servidores de Model Context Protocol (MCP) directamente en el IDE.

Esta arquitectura habilita a los modelos de lenguaje (LLMs) a interactuar con los metadatos de los catálogos empresariales, ejecutar scripts de validación, gestionar el ciclo de vida de los datos y proponer planes de implementación detallados sin comprometer la seguridad ni los límites de gobernanza del proyecto.

Matriz de Decisión Arquitectónica: Estrategias de Transformación y Análisis

La selección del motor de ejecución determina el rendimiento y el control de costes en la nube. A continuación, se comparan los patrones más habituales en arquitecturas modernas de datos:

Implementación Práctica: Infraestructura, Pipeline ETL y Agentes

A continuación se detalla la arquitectura completa de tres componentes: aprovisionamiento IaC con Terraform, ingestión/limpieza mediante BigFrames y explotación con el Agente.

1. Definición de Infraestructura con Terraform (GCS + BigQuery)

Creamos el bucket de almacenamiento crudo y el dataset analítico (Gold Layer) en la región europe-southwest1 (Madrid):

# main.tf
terraform {
  required_version = ">= 1.5.0"
  required_providers {
    google = {
      source  = "hashicorp/google"
      version = "~> 5.0"
    }
  }
}

provider "google" {
  project = var.project_id
  region  = var.region
}

# Bucket para la capa Bronze / Raw
resource "google_storage_bucket" "raw_bucket" {
  name                        = "${var.project_id}-raw-data-bucket"
  location                    = var.region
  force_destroy               = true
  uniform_bucket_level_access = true
  public_access_prevention    = "enforced"
}

# Dataset para la capa Gold
resource "google_bigquery_dataset" "gold_dataset" {
  dataset_id  = "gold_layer"
  description = "Almacena los datos analiticos limpios y listos para consumo"
  location    = var.region
}

2. Script de Generación y Carga de Transacciones Sintéticas

# generator.py
import json
import random
from faker import Faker
from google.cloud import storage

fake = Faker()

def generate_synthetic_transactions(num_records: int = 2000):
    transactions = []
    for _ in range(num_records):
        transactions.append({
            "transaction_id": fake.uuid4(),
            "customer_name": fake.name(),
            "email": fake.email(),
            # Simula valores negativos/atipicos para la etapa de limpieza
            "amount": round(random.uniform(-50.0, 1500.0), 2),
            "country": fake.country() if random.random() > 0.1 else None,
            "timestamp": fake.date_time_this_year().isoformat()
        })
    return transactions

def upload_to_gcs(bucket_name: str, destination_blob: str, data: list):
    client = storage.Client()
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(destination_blob)
    blob.upload_from_string(
        data=json.dumps(data),
        content_type="application/json"
    )
    print(f"Dataset subido exitosamente a gs://{bucket_name}/{destination_blob}")

if __name__ == "__main__":
    PROJECT_ID = "curso-gcp-data-agent-kit"
    BUCKET_NAME = f"{PROJECT_ID}-raw-data-bucket"
    records = generate_synthetic_transactions(2000)
    upload_to_gcs(BUCKET_NAME, "raw_transactions.json", records)

3. Pipeline de Limpieza y Persistencia con BigQuery DataFrames

# data_pipeline.py
import bigframes.pandas as bpd
from google.cloud import storage
import pandas as pd
import json

PROJECT_ID = "curso-gcp-data-agent-kit"
LOCATION = "europe-southwest1"
BUCKET_NAME = f"{PROJECT_ID}-raw-data-bucket"
BLOB_NAME = "raw_transactions.json"

# Inicializacion de BigQuery DataFrames
bpd.options.bigquery.project = PROJECT_ID
bpd.options.bigquery.location = LOCATION

# 1. Lectura del origen crudo desde Cloud Storage
storage_client = storage.Client(project=PROJECT_ID)
bucket = storage_client.bucket(BUCKET_NAME)
blob = bucket.blob(BLOB_NAME)
raw_data = json.loads(blob.download_as_text())

# Convertir a DataFrame inicial e instanciar en BigFrames
pdf_raw = pd.DataFrame(raw_data)
df = bpd.read_pandas(pdf_raw)

# 2. Reglas de Calidad de Datos (Pushdown Execution)
# - Eliminar transacciones sin pais registrado
# - Filtrar unicamente montos transaccionales estrictamente positivos
df_cleaned = df.dropna(subset=["country"])
df_cleaned = df_cleaned[df_cleaned["amount"] > 0]

# 3. Persistencia en la Capa Gold de BigQuery
destination_table = f"{PROJECT_ID}.gold_layer.clean_transactions"
df_cleaned.to_gbq(destination_table, if_exists="replace")

print(f"Pipeline completado. Tabla disponible en: {destination_table}")

Framework de Adopción: Guía Paso a Paso

1. Creación de Proyecto y Autenticación ADC: Crea un proyecto en GCP, habilita la facturación y ejecuta gcloud auth login y gcloud auth application-default login en la terminal local para sincronizar credenciales.
2. Instalación del Data Agent Kit: Abre Visual Studio Code o Cursor/Antigravity, dirígete al Marketplace de extensiones e instala Google Cloud Data Agent Kit.
3. Configuración de MCP Servers y APIs: Entra en la pestaña de configuración del agente (Quick Start Guide), enlaza tu Project ID y región predeterminada (europe-southwest1), y activa las APIs analíticas necesarias (BigQuery, Storage, Dataproc, Gemini Data Analytics).
4. Despliegue de Recursos con Terraform: Inicializa el proveedor (terraform init) y aplica el plan (terraform apply -auto-approve) para construir el bucket y el dataset de destino.
5. Generación de Lógica Asistida por Agentes: Mediante el panel de chat del agente, solicita la creación del plan de implementación en Markdown y genera el notebook con bigframes para aplicar las reglas de negocio.
6. Explotación con Conversational Analytics: Desde el explorador de BigQuery del agente, selecciona la tabla clean_transactions y abre la interfaz de chat interactivo para calcular agregaciones, tendencias y distribuciones por país sin escribir SQL manual.