Canalización de detección de fraudes con Data Agent Kit y el IDE de Antigravity

1. Introducción

Imagina que eres un científico de datos en Cymbal Financial, un procesador de pagos de gran volumen. Se produjo una ola de retrasos en la liquidación, y el equipo de cumplimiento sospecha de fraude coordinado. Debes compilar una canalización para transferir registros sin procesar de transacciones de la cámara de compensación, limpiar los datos, entrenar un modelo de aprendizaje automático, ejecutar la inferencia por lotes y transferir las transacciones de alto riesgo a una cola de revisión de Cloud Spanner para la auditoría manual.

Normalmente, esto requiere días de escribir código de configuración repetitivo (notebooks de Spark, configuraciones de dbt, secuencias de comandos de entrenamiento, DAGs de Airflow) y un cambio constante de contexto entre las interfaces de la consola y los editores.

En este codelab, programarás en pareja con un agente usando el Kit de agente de datos (DAK) de Google Cloud dentro del IDE de Antigravity. Con lenguaje natural conversacional, el agente te ayudará a generar notebooks de Spark, compilar un proyecto de dbt, construir un bucle de inferencia y organizar el flujo de trabajo con Managed Service para Apache Airflow.

Actividades

• Transfiere registros de la cámara de compensación desde Cloud Storage con Managed Service para Apache Spark (Spark sin servidores) a una tabla de BigQuery.
• Anula duplicados y normaliza las transacciones con dbt para establecer capas de datos limpios (sin procesar, de etapa de pruebas y enriquecidos).
• Entrena un modelo de clasificación de bosque aleatorio distribuido (RandomForestClassifier) en Spark sin servidores.
• Ejecuta la inferencia por lotes en transacciones nuevas y escribe alertas de alto riesgo directamente en Cloud Spanner.
• Organiza, configura visualmente y, luego, implementa toda la canalización con Managed Service para Apache Airflow y la supervisión interactiva de DAG dentro del IDE.

Requisitos

• Un navegador web, como Chrome
• Un proyecto de Google Cloud con la facturación habilitada (te recomendamos que uses un proyecto nuevo y dedicado para los labs prácticos).
• Conocimientos básicos de SQL, Python y PySpark
• IDE de Antigravity con una suscripción a Google AI Pro (recomendado)

Los recursos creados en este codelab deberían costar menos de USD 5. Asegúrate de seguir las instrucciones de Clean Up al final del lab para borrar los recursos aprovisionados.

Elige tu herramienta: En este lab, se usa el IDE de Antigravity (la extensión funciona en todas las bifurcaciones de VS Code). Pronto lanzaremos versiones alternativas para VS Code, así como flujos de trabajo basados en complementos para Antigravity 2.0, la CLI de Antigravity y Claude Code.

2. Configuración del entorno

Para comenzar el lab, ejecutarás una secuencia de comandos de arranque. Esta secuencia de comandos habilita automáticamente las APIs de GCP requeridas, crea un bucket de Cloud Storage de transferencia, genera conjuntos de datos de directorio y transacciones simulados, carga directorios de referencia en BigQuery y activa el aprovisionamiento en segundo plano de Cloud Spanner y Managed Service para Apache Airflow (antes conocido como Cloud Composer).

Selecciona o crea un proyecto

Elige un proyecto existente o crea uno nuevo en la consola de Google Cloud.

Verifica la facturación

Asegúrate de que la facturación esté habilitada para tu proyecto de Google Cloud. Sigue esta guía para obtener más información sobre cómo hacerlo.

Ejecuta la secuencia de comandos de configuración

Usarás Google Cloud Shell (o tu shell local configurado con Google Cloud CLI) para iniciar la configuración del entorno.

1. Abre la consola de Google Cloud.
2. Haz clic en Activar Cloud Shell en la barra de herramientas superior derecha.

3. En la terminal de Cloud Shell, configura tu proyecto activo:

gcloud config set project <<YOUR_PROJECT_ID>>
export PROJECT_ID=$(gcloud config get-value project)

4. Clona el repositorio del codelab y navega a la carpeta de secuencias de comandos:

cd ~/
git clone --filter=blob:none --no-checkout https://github.com/GoogleCloudPlatform/devrel-demos.git
cd ~/devrel-demos
git sparse-checkout init --cone
git sparse-checkout set codelabs/agentic-data-labs/data-science
git checkout main
cd codelabs/agentic-data-labs/data-science/scripts

5. Ejecuta la secuencia de comandos de configuración de inicio para implementar todos los recursos en us-central1:

chmod +x setup.sh setup_spanner.sh setup_composer.sh
export REGION=us-central1
./setup.sh

Qué hace setup.sh:

• Habilita las APIs: Activa las APIs de BigQuery, Dataproc, Spanner, Managed Airflow, Cloud Storage y las de infraestructura de asistencia.
• Crea buckets de Cloud Storage: Provisiona buckets para los registros de transacciones sin procesar y los artefactos del modelo de AA.
• Almacena datos iniciales: Copia los registros de transacciones sin procesar en Cloud Storage y carga los datos de referencia en BigQuery.
• Configura Spark Serverless: Crea la plantilla de sesión del entorno de ejecución con el conector de Cloud Spanner.
• Aprovisiona servicios en segundo plano: Genera trabajadores en segundo plano para compilar Cloud Spanner y Managed Airflow.
• Genera .env: Guarda la configuración del entorno para los codelabs y las secuencias de comandos de limpieza.

Los recursos creados por la secuencia de comandos de configuración se borran en el paso final de limpieza con ./teardown.sh.

6. Cuando finalice la secuencia de comandos, verás un resumen que indica que tu conjunto de datos de BigQuery y tu bucket de Cloud Storage están listos. En segundo plano, Cloud Spanner (tarda alrededor de 2 minutos) y Managed Airflow (tarda alrededor de 20 minutos) seguirán aprovisionándose. Puedes supervisar su progreso en cualquier momento ejecutando el siguiente comando:

tail -f /tmp/spanner_setup.log
tail -f /tmp/composer_setup.log

Ciclo de desarrollo optimizado: Dado que el aprovisionamiento de Managed Airflow tarda alrededor de 20 minutos, la secuencia de comandos de configuración lo inicia inmediatamente en segundo plano. Esto te permite trabajar en la página 3 (Ingestion), la página 4 (dbt Transformations), la página 5 (ML Training) y la página 6 (Batch Inference) mientras se inicia Airflow. Cuando llegues a la página 7 (Orquestación), tu entorno estará completamente listo.

Abre el IDE de Antigravity

1. Descarga e instala el IDE de Antigravity desde la página de descarga de Google Antigravity.
2. Inicia el IDE de Antigravity.
3. Crea una carpeta nueva y vacía en tu máquina local (p. ej., llamada agentic-data-labs) y ábrela en el IDE. Para ello, elige Abrir carpeta. Este será tu espacio de trabajo local para el codelab.

Instala la extensión de Data Agent Kit

La extensión Google Cloud Data Agent Kit proporciona una integración profunda con los servicios de datos de Google Cloud directamente en tu editor, lo que te permite interactuar con BigQuery, Cloud SQL, Cloud Storage y muchos más sin cambiar de contexto.

1. En el IDE de Antigravity, haz clic en el ícono de Extensiones en la barra de actividad, en el extremo izquierdo de la pantalla (parece cuatro cuadrados).
2. En la barra de búsqueda que se encuentra en la parte superior del panel Extensiones, escribe Google Cloud Data Agent Kit.
3. Busca la extensión llamada Google Cloud Data Agent Kit publicada por googlecloudtools.
4. Haz clic en el botón Install.
5. Es posible que aparezca un mensaje que pregunte: "¿Confías en el publicador "googlecloudtools" y sus extensiones?". Haz clic en Confiar en los publicadores y, luego, en Instalar para continuar.

Una vez instalado, verás un nuevo ícono de Google Cloud Data Agent Kit en la barra de actividad, en el extremo izquierdo del IDE de Antigravity.

6. Se abrirá automáticamente una página de incorporación titulada "Te damos la bienvenida al kit de agentes de datos de Google Cloud". Si no accediste a tu cuenta de Cloud, sigue las instrucciones para permitir el acceso.
7. En la sección Resumen de la configuración, busca el campo del proyecto. Haz clic en el menú desplegable y selecciona tu proyecto de Google Cloud. Establece tu región como us-central1. Luego, selecciona Configure MCP Servers.

8. Selecciona Configurar servidores de MCP. En el panel Configuración de MCP, asegúrate de habilitar los siguientes servidores de MCP remotos:
   • BigQuery
   • Spanner
   • Notebooks

Luego, haz clic en Comenzar.

¿Ya instalaste el complemento o no se abrió la página de bienvenida? Puedes abrir la configuración de la extensión manualmente en cualquier momento. Haz clic en el ícono de Google Cloud Data Agent Kit en la barra de actividades que se encuentra en el extremo izquierdo y, luego, selecciona Configuración para activar o desactivar tu configuración.

Nota sobre las habilidades o los MCP activos: Si tienes otras habilidades personalizadas o servidores de MCP activos en tu IDE, es posible que interfieran en el flujo esperado de este lab. Si el agente se comporta de forma inesperada, considera inhabilitarlo para este lab.

Explora las opciones de configuración

Una vez que se complete la configuración, accederás a la página "Comienza a usar el kit de agentes de datos de Google Cloud".

1. En "Configuración", haz clic en Comenzar.
2. Se abrirá el panel Configuración del kit del agente de datos. Explora las pestañas:
   • Proyecto y región: Verifica el ID del proyecto seleccionado y confirma que la secuencia de comandos de configuración habilitó todas las APIs necesarias (Compute Engine, Cloud Storage, BigQuery, Spanner, etcétera).
   • BigQuery: Configura la ubicación predeterminada para tus consultas de BigQuery. Usa la región us-central1.
   • Configura servidores de MCP: Consulta los servidores de MCP habilitados (BigQuery, Notebooks, Spanner, etc.) que permiten que los agentes de IA interactúen de forma segura con tus datos.
   • Habilidades: Explora las habilidades prediseñadas que proporcionan a los agentes capacidades especializadas para tareas de datos complejas.

Resumen de la sección: Ejecutaste la secuencia de comandos de arranque para crear recursos de GCS y BigQuery mientras Spanner y Airflow se compilan en segundo plano. Luego, abriste el proyecto en el IDE de Antigravity y activaste la extensión del kit de agentes de datos de Google Cloud. Ya puedes escribir tu primer notebook.

3. Ingiere registros sin procesar con Spark sin servidores

En esta sección, transferirás registros de transacciones JSON sin procesar al data lake. Managed Service para Apache Spark (Spark sin servidores) se conecta directamente con el almacenamiento nativo de BigQuery. Usarás el conector estándar de BigQuery para administrar datos tabulares y habilitar las consultas y el análisis directos.

Explora el entorno de ejecución de Spark sin servidores previamente configurado

Antes de ejecutar el código de Spark, inspecciona la plantilla del entorno de ejecución sin servidor que preconfiguró la secuencia de comandos de configuración. Esta plantilla define el backend del entorno de ejecución de destino y agrupa las dependencias del conector necesarias.

1. En la barra de actividad del IDE, abre el panel Google Cloud Data Agent Kit.
2. Expande el menú desplegable Apache Spark y, luego, Serverless.
3. Haz clic con el botón derecho en fraud-pipeline-runtime y selecciona Profile para abrir su vista de configuración en el editor.
4. En la pestaña Perfil, desplázate hacia abajo y expande Propiedades para inspeccionar las dependencias personalizadas adjuntas al entorno:
   • spark.jars: Contiene gs://spark-lib/spanner/spark-3.5-spanner-1.4.0.jar, que usa el conector de Spark Spanner para permitir que los trabajos de Spark escriban los resultados de la inferencia directamente en Cloud Spanner más adelante en el lab. (Nota: Dataproc Serverless incluye el conector de Spark BigQuery de Google Cloud de forma predeterminada, por lo que no se requiere ninguna configuración adicional de JAR para leer y escribir tablas de BigQuery).

5. Observa la pestaña Sesiones interactivas a la izquierda. Actualmente, está vacío porque aún no ejecutaste ningún código. En cuanto ejecutes el notebook en el siguiente paso, se aprovisionará dinámicamente una sesión de procesamiento sin servidores activa y aparecerá aquí.

Transfiere datos con el Data Agent Kit

En lugar de configurar manualmente una sesión de Spark o escribir scripts de carga de PySpark desde cero, programarás en pareja con un agente usando Data Agent Kit.

1. Abre el panel Agent Chat haciendo clic en el ícono Toggle Agent en la barra de herramientas de la esquina superior derecha.
2. Pega la siguiente instrucción en el chat (asegúrate de reemplazar ${PROJECT_ID} por el ID de tu proyecto de Google Cloud):

Create a PySpark notebook (01_ingestion.ipynb) to ingest JSON transaction logs
from gs://${PROJECT_ID}-fin-clearing-raw/ into a BigQuery table
`${PROJECT_ID}.transactions_dataset_evals.raw_transactions`
using the Spark BigQuery connector (`format("bigquery")`) with overwrite mode.

3. Si el agente solicita permiso para ejecutar comandos de verificación en segundo plano (p.ej., "¿Permitir ejecutar este comando?"), revisa el comando propuesto y selecciona Sí, permitir esta vez (o Sí, y permitir siempre).
4. Cuando el agente termine de generar el archivo, haz clic en el botón azul Aceptar todo (o en el ícono de marca de verificación) que se encuentra en la parte inferior del panel de chat para guardar notebooks/01_ingestion.ipynb en tu espacio de trabajo.

Aprobaciones de comandos y aceptación de cambios: Como programador par de IA, el agente puede solicitar permiso para ejecutar comandos seguros de la terminal (como verificaciones de sintaxis o inspección de dependencias) antes de generar código. Revisa y permite estos comandos cuando se te solicite, y haz clic en Continuar o Aceptar todo en el panel de chat cada vez que se presenten archivos o planes de implementación.

Revisa y ejecuta el notebook

1. Abre el archivo notebooks/01_ingestion.ipynb recién generado en el IDE.
2. Revisa el código de PySpark para la lógica de escritura del conector de BigQuery.
3. Haz clic en Run All en la barra de herramientas del notebook del IDE.
4. Si es la primera vez que ejecutas un notebook de Spark remoto, es posible que el IDE te solicite que instales dependencias locales. Si se te solicita, haz clic en Install dependencies for Remote Spark Kernels, confirma los diálogos de instalación y, luego, vuelve a hacer clic en Run All.
5. En el menú desplegable Select Kernel, elige Remote Spark Kernels -> fraud-pipeline-runtime on Serverless Spark. (Sugerencia: Si no ves la plantilla de tiempo de ejecución preconfigurada en la lista, haz clic en el ícono de actualización en la esquina superior derecha del menú desplegable del selector de kernel para volver a cargar los kernels remotos disponibles).
6. Mira la barra de estado en la parte inferior izquierda del editor. Verás Connecting to kernel: fraud-pipeline-runtime on Serverless Spark.... Dado que este es el lanzamiento inicial del backend del kernel del entorno de ejecución de Spark sin servidores, tardará unos minutos en aprovisionarse y arrancar.
7. Una vez que el kernel termine de conectarse, el notebook comenzará automáticamente a ejecutar todas las celdas de forma secuencial para procesar los registros de transacciones sin procesar en tu conjunto de datos de BigQuery.

Conexión a Spark Serverless: Cuando te conectes a la sesión de tiempo de ejecución de Spark Serverless, verás Connecting to kernel... en la barra de estado inferior. La inicialización de los recursos en Google Cloud tarda alrededor de 3 minutos antes de que el notebook se conecte y ejecute todas las celdas.

Verificación

Una vez que se complete la ejecución, consulta el catálogo de Data Agent Kit para verificar la creación de la tabla:

1. En la barra de actividad del IDE, abre el panel Google Cloud Data Agent Kit.
2. Expande la sección CATALOG.
3. Expande tu ID del proyecto.
4. Expande BigQuery.
5. Expande el conjunto de datos transactions_dataset_evals.
6. Haz clic en la tabla raw_transactions para abrir su vista de detalles en el editor principal.
7. En la navegación de la izquierda, explora las pestañas Datos, Esquema y Detalles para inspeccionar los registros y los metadatos que se transfirieron.

Resumen de la sección: Usaste lenguaje natural en el chat del agente para generar una carga de trabajo completa de Spark sin servidores. Luego, lo ejecutaste para procesar registros JSON no estructurados en una tabla (sin procesar) de BigQuery.

4. Anula duplicados y normaliza con dbt

Antes de entrenar el modelo de AA, aplicarás la calidad de los datos quitando los registros de transmisión duplicados, aislando los registros incorrectos (como los IDs de transacción vacíos) y uniendo los datos dimensionales (pagadores y beneficiarios). Este proceso requiere transformaciones de SQL idempotentes y confiables, lo que hace que dbt (data build tool) sea una excelente opción.

Crea el esqueleto de la canalización de dbt

Usa el agente para generar un proyecto de dbt sobre el conjunto de datos de BigQuery:

1. Vuelve al panel Agent Chat.
2. Proporciona la siguiente instrucción para generar el proyecto de dbt:

Scaffold a us-central1 dbt project in dbt_project/ that maps raw_transactions
through to an enriched_transactions model in dataset transactions_dataset_evals.

Deduplicate by transaction_id in staging. Quarantine null IDs to an invalid_transactions model.
Join the valid staging records with dim_payers and dim_payees for the enriched_transactions model,
preserving the historical `is_fraud` label column, and finally add a transaction uniqueness test.

Create an implementation plan first.

3. El agente presentará un artefacto de Plan de implementación en el panel principal del editor. Revisa la estructura de archivos y la lógica de SQL propuestas.
4. Haz clic en Continuar (y, luego, en Aceptar todo) para permitir que el agente genere los archivos en tu espacio de trabajo.

5. Una vez que se complete la generación, el agente mostrará un Recorrido que resume los componentes nuevos. Acepta todos los cambios si se te solicita.

Administración inteligente de dependencias: Cuando se crean proyectos de scaffolding, el agente inspecciona de forma proactiva tu entorno de shell en busca de las herramientas de la CLI necesarias (como dbt). Si faltan, el agente puede ofrecer automáticamente crear un entorno virtual e instalar dbt-bigquery en segundo plano, y solicitar tu aprobación antes de ejecutar los comandos de configuración.

Compilación y prueba

Si bien el agente ejecutó dbt compile automáticamente para garantizar que el código SQL generado fuera sintácticamente válido, ahora materializarás estas vistas y tablas en BigQuery y ejecutarás las pruebas de calidad de los datos para la verificación local. (Nota: Más adelante en el lab, automatizarás este paso de dbt como parte de un DAG de Airflow de extremo a extremo).

1. En la barra de actividad que se encuentra en el extremo izquierdo, haz clic en el ícono Explorador (o presiona Cmd/Ctrl+Shift+E).
2. Expande dbt_project -> models para inspeccionar los modelos de SQL generados. Haz clic en enriched_transactions.sql para abrir y revisar la lógica de la transformación y la función de fraude en el editor.
3. En el Explorador de archivos, haz clic con el botón derecho en la carpeta dbt_project y selecciona Abrir en la terminal integrada. Esto abre automáticamente un panel de terminal configurado directamente en el directorio de trabajo dbt_project requerido.
4. Si aún no tienes instalado dbt, crea un entorno virtual fuera de dbt_project/ (en la raíz de tu directorio principal o espacio de trabajo) y, luego, instala el adaptador de BigQuery:

python3 -m venv ~/.venv/dbt
source ~/.venv/dbt/bin/activate
pip install dbt-bigquery

5. Ejecuta los modelos de dbt y sus pruebas de calidad de los datos asociadas:

dbt build

6. Observa el resultado de la terminal. dbt compilará el código SQL, materializará las tablas de etapa de pruebas y enriquecidas en BigQuery, y ejecutará las pruebas de datos.

Solución de problemas de compilaciones con el agente: Si dbt build encuentra algún error (como discrepancias de esquemas o relaciones faltantes), copia y pega el resultado del error directamente en el chat del agente. El agente analizará la falla, ajustará los modelos y resolverá el problema por ti.

7. Una vez que finalice la compilación, cierra el panel de la terminal para liberar espacio en la pantalla para los pasos restantes.

Resumen de la sección: Generaste un proyecto de dbt con el agente, ejecutaste pruebas de calidad de los datos y transformaste los registros sin procesar en tablas de BigQuery enriquecidas y de etapa de pruebas.

5. Entrena un modelo de detección de fraudes distribuido con Random Forest

Con las transacciones enriquecidas materializadas en BigQuery, crearás un modelo de aprendizaje automático para clasificar los eventos fraudulentos. Random Forest es un método de aprendizaje ensamblado adecuado para los datos de clasificación tabulares. Ejecutar un RandomForestClassifier en Spark Serverless distribuye el entrenamiento del modelo en nodos de trabajo sin necesidad de que administres la infraestructura.

En este paso, usarás el agente para generar la canalización de entrenamiento de AA de Spark.

Genera el notebook de entrenamiento del AA

1. Abre el panel Agent Chat.
2. Proporciona la siguiente instrucción para diseñar la secuencia de entrenamiento del modelo (recuerda reemplazar ${PROJECT_ID} por el ID de tu proyecto activo):

Create a PySpark notebook (02_training.ipynb) to train a distributed Random Forest
(RandomForestClassifier) model on the BigQuery table
`transactions_dataset_evals`.`enriched_transactions`, predicting the `is_fraud` label.
Train only on historically labeled records where `is_fraud` is not null.

One-hot encode categorical strings, scale amounts, cache the dataset in memory,
evaluate AUC, and save the evaluated model to gs://${PROJECT_ID}-models/fraud_model.

3. Revisa el plan del agente o el código generado y haz clic en Continuar o Aceptar todo para guardar notebooks/02_training.ipynb en tu espacio de trabajo.

Revisa y ejecuta el notebook

1. Abre notebooks/02_training.ipynb en el editor.
2. Revisa las etapas de la canalización de AA de PySpark para la codificación de atributos, el ensamblaje de vectores y la lógica de clasificación de Random Forest.
3. Haz clic en Run All en la barra de herramientas del notebook del IDE.
4. Cuando se abra el selector desplegable Select Kernel, selecciona fraud-pipeline-runtime on Serverless Spark.

Verificación

Una vez que se complete la ejecución, confirma que el modelo se entrenó y exportó correctamente:

1. Revisa los resultados de las celdas de evaluación cerca de la parte inferior del notebook para verificar la puntuación del área bajo la curva ROC (AUC) informada.
2. Para asegurarte de que los artefactos del modelo se guardaron correctamente en GCS, expande el panel del explorador STORAGE en la barra lateral de Data Agent Kit.
3. Ubica el bucket que termina en -models (vinculado a tu ID de proyecto activo), expándelo y explora en detalle para verificar que existan el directorio fraud_model y sus etapas de canalización.

Del notebook a la producción: Si bien los notebooks interactivos son excelentes para la experimentación y el desarrollo de modelos, tu agente también puede refactorizar el código del notebook en secuencias de comandos de PySpark independientes y parametrizables para la ejecución recurrente por lotes en Dataproc Serverless.

Resumen de la sección: Usaste el agente para crear una canalización de entrenamiento de AA de PySpark, entrenaste un modelo de Random Forest en tu tabla enriquecida de BigQuery y exportaste el modelo a Cloud Storage.

6. Inferencia por lotes y escritura en Cloud Spanner

Con un modelo predictivo entrenado almacenado en Cloud Storage, ejecutarás la inferencia por lotes en las transacciones nuevas que fluyen a través de BigQuery. Las transacciones de alto riesgo deben dirigirse a un sistema operativo para que un equipo de cumplimiento pueda revisarlas. Cloud Spanner proporciona una base de datos transaccional escalable para esta cola de revisión.

Genera el notebook de inferencia por lotes

Usa el agente para crear un notebook de inferencia que conecte BigQuery, Cloud Storage y Cloud Spanner:

1. Abre el panel Agent Chat.
2. Proporciona la siguiente instrucción:

Create an inference notebook (03_inference.ipynb) that loads the RandomForestClassifier
model to score unlabeled records (where `is_fraud` is null) from the BigQuery table
`transactions_dataset_evals`.`enriched_transactions`.

Filter for high-risk transactions with a 50%+ fraud probability score (probability >= 0.50)
and write them to the Cloud Spanner table SparkEvalFraudReviewQueue in the cymbal-fraud instance
under fraud-db.

3. Acepta el notebook generado para guardar notebooks/03_inference.ipynb en tu espacio de trabajo.

Por qué el 50% se considera de riesgo alto: Si bien el 50% parece un lanzamiento de moneda, la gran mayoría de las transacciones legítimas obtienen una puntuación inferior al 5%. Una puntuación superior al 50%representa el 2% de las transacciones más anómalas de la red, lo que filtra el ruido para que los auditores solo revisen los eventos genuinos de alto riesgo.

Revisa y ejecuta el notebook

1. Abre el archivo notebooks/03_inference.ipynb recién generado en el editor.
2. Revisa la secuencia de inferencia de PySpark:
   • Dependencias: La plantilla de entorno de ejecución sin servidores proporciona las dependencias JAR cloud-spanner necesarias para la ejecución de Spark.
   • Formato de datos: La secuencia de comandos descarta las columnas vectoriales complejas de AA de Spark (como las probabilidades y las características sin procesar) antes de escribir para que coincidan con el esquema de la tabla de Spanner.
   • Spanner Connector: Escribe las filas marcadas con .format("cloud-spanner") para agregarlas directamente a la cola de revisión.
3. Haz clic en Run All en la barra de herramientas del notebook del IDE.
4. Cuando se te solicite que selecciones un kernel, elige fraud-pipeline-runtime on Serverless Spark.

Verificación

Una vez que el notebook de inferencia termine de procesarse, podrás consultar tu base de datos operativa de Spanner directamente en el IDE:

1. En la barra de actividad del IDE, abre el panel Google Cloud Data Agent Kit.
2. Expande la sección CATALOG.
3. Expande tu ID del proyecto y, luego, expande Spanner.
4. Navega a cymbal-fraud -> fraud-db -> Tables -> SparkEvalFraudReviewQueue.
5. Haz clic con el botón derecho en la tabla y selecciona Query Table. Luego, ejecuta la consulta:

SELECT *
FROM `SparkEvalFraudReviewQueue`
LIMIT 100;

6. En el panel Resultados de la consulta que se encuentra a continuación, deberías ver las filas recién insertadas que representan transacciones de alto riesgo marcadas para su revisión manual.

Resumen de la sección: Usaste el agente para crear un notebook de inferencia por lotes, calificaste registros de BigQuery sin etiquetar con tu modelo entrenado y escribiste transacciones de alto riesgo directamente en Cloud Spanner.

7. Crea estructuras y coordina con Managed Airflow

Actualmente, tu canalización consta de pasos discretos: un notebook de transferencia, un proyecto de transformación de dbt y un notebook de inferencia por lotes. Para que esté lista para producción, las unirás en un gráfico de dependencias programado.

Managed Service para Apache Airflow (antes conocido como Cloud Composer) proporciona un motor de organización administrado para este flujo de trabajo. El Data Agent Kit incluye una función de Canalizaciones de organización que traduce las definiciones de canalizaciones declarativas de YAML directamente en DAG de Airflow.

Define la canalización

Usa el agente para generar la configuración de la canalización de organización:

1. En el Chat del agente, proporciona la siguiente instrucción (recuerda reemplazar ${PROJECT_ID}):

Initialize and define an orchestration pipeline (fraud_analysis_pipeline) triggering
the ingestion notebook, dbt project, and inference notebook in sequential order.
For Dataproc Serverless engine configs in us-central1, use resourceProfile.inline
(defining properties with spark.jars: "gs://spark-lib/spanner/spark-3.5-spanner-1.4.0.jar"
for inference) rather than resourceProfile.path or overrides.

Set the schedule interval to run daily at midnight, and use
gs://${PROJECT_ID}-airflow-artifacts for artifact storage.

Revisa la configuración del DAG

El organizador de Data Agent Kit usa parámetros de configuración YAML declarativos para definir y, luego, implementar canalizaciones en Apache Airflow, lo que permite que las definiciones se controlen por versión y se implementen a través de CI/CD.

En el panel IDE Explorer, revisa los dos archivos de canalización que generó el agente en la raíz de tu espacio de trabajo:

1. deployment.yaml: Abre este archivo. Esto sirve como registro de tu entorno. Asigna tu canalización lógica dev al entorno cymbal-airflow, establece la región de ejecución (us-central1) y define el bucket artifact_storage en el que se almacenan los DAG compilados y las dependencias.
2. fraud_analysis_pipeline.yaml: Abre este archivo. Esto define el gráfico de ejecución. Especifica el programa del activador (interval: '0 0 * * *') y secuencia los tres pasos en el bloque actions:
   • Una acción de notebook de transferencia para 01_ingestion.ipynb que se ejecuta en Dataproc Serverless.
   • Una acción de transformación pipeline que segmenta el directorio dbt_project, con una dependencia dependsOn que apunta al paso de transferencia.
   • Es una acción de inferencia notebook para 03_inference.ipynb con una dependencia dependsOn que apunta al paso de dbt y que incluye la propiedad del JAR de Spanner.
3. El agente también resumirá estos artefactos generados en una pestaña Recorrido en el panel del editor, en la que se describirán las configuraciones y las validaciones realizadas.

Configuración interactiva del DAG

El Data Agent Kit renderiza la configuración de tu canalización como un gráfico visual interactivo para inspeccionar y editar las propiedades del DAG de Airflow.

1. En la barra de actividad del IDE, abre el panel Google Cloud Data Agent Kit.
2. En DATA ENGINEERING, expande Orchestration Pipelines.
3. Haz clic en fraud_analysis_pipeline.yaml para abrir el lienzo del DAG visual en el editor principal.

4. Haz clic en el nodo Schedule trigger que se encuentra en la parte superior. Se abrirá un panel desplegable de configuración a la derecha, en el que se mostrará la cadena de Cron analizada (0 0 * * *) y se te permitirá ajustar parámetros como el reabastecimiento y la recuperación.
5. Haz clic en cualquiera de los nodos de tareas de notebook (como el paso de inferencia o de transferencia). La ventana emergente se actualiza para mostrar las asignaciones de ejecución y las propiedades del conector específicas de Dataproc Serverless.
6. Observa el hipervínculo del nombre de archivo del notebook (como 01_ingestion.ipynb) dentro del bloque del nodo. Si haces clic en él, se abrirá el notebook directamente en el editor.
7. En la barra lateral izquierda, debajo de Canalizaciones de organización, haz clic en Deployment configuration. En esta vista, se muestran el clúster del entorno de dev de destino y los artefactos del bucket de GCS de salida.

Resumen de la sección: Generaste una configuración de canalización de organización con el agente, que define las dependencias entre las tareas de inferencia, dbt y la transferencia en un lienzo visual interactivo.

8. Implementa, ejecuta y supervisa

Con el DAG definido de forma local, te conectarás al entorno de Managed Airflow aprovisionado durante la configuración y, luego, implementarás la canalización.

Configura Managed Service para Apache Airflow

Antes de la implementación, configura la conexión del planificador en los parámetros de configuración del Data Agent Kit para que la extensión apunte a tu entorno de Managed Airflow:

1. En la barra de actividad del IDE, abre el panel Google Cloud Data Agent Kit.
2. En SETTINGS, haz clic en Configuración.
3. Selecciona Programador en el menú de la izquierda.
4. Configura los parámetros:
   • ID del proyecto: Selecciona el ID de tu proyecto activo.
   • Región: Selecciona us-central1.
   • Entorno: Selecciona cymbal-airflow.
5. Haz clic en Guardar.

Implementa el DAG

Optimiza los artefactos de implementación: Antes de implementar, asegúrate de que las carpetas temporales, como .venv, target/ o logs/, no estén anidadas dentro de tu directorio dbt_project/. Subir entornos virtuales grandes que contienen miles de archivos puede hacer que el empaquetado se detenga y que se ralentice el análisis de DAG de Airflow. Si están presentes, quítalos (rm -rf dbt_project/.venv dbt_project/target dbt_project/logs) o agrégalos a un archivo .airflowignore.

Ahora implementarás la canalización configurada directamente en tu entorno de Managed Airflow desde el lienzo visual:

1. En la barra lateral del Google Cloud Data Agent Kit, expande DATA ENGINEERING > Orchestration Pipelines y haz clic en fraud_analysis_pipeline.yaml para abrir el lienzo visual del DAG.
2. En la esquina superior derecha de la barra de herramientas del lienzo, haz clic en el botón azul Ejecutar canalización.
3. En el selector del menú desplegable del entorno, selecciona dev.
4. Observa la notificación de progreso en el área de estado inferior (Running pipeline: Building pipeline locally...). La extensión compilará automáticamente tu DAG, empaquetará el notebook y los recursos de dbt, y los subirá al bucket de GCS de tu entorno de Managed Airflow (este proceso tarda entre 3 y 4 minutos en completarse).

Supervisa la ejecución

Una vez que se complete la compilación local y la notificación emergente confirme Triggered a new run for pipeline... successfully, supervisa la ejecución en vivo:

1. En la barra lateral Google Cloud Data Agent Kit, expande DATA ENGINEERING > Orchestration Pipelines.
2. Haz clic en Administración de canalizaciones.
3. En la tabla Pipelines Management, haz clic en fraud_analysis_pipeline para abrir el historial de ejecución.

4. En la vista Historial de ejecución, selecciona la ejecución activa en el calendario.
5. A medida que avanza la ejecución en cada tarea de la canalización (transferencia, transformación de dbt y la inferencia), se actualizan los indicadores de estado y se completan las duraciones de las tareas. Haz clic en cualquier tarea para inspeccionar su resultado de la ejecución en vivo y los registros del DAG de Airflow.

Implementaciones de producción con CI/CD y GitHub Actions: Además de las implementaciones locales, Data Agent Kit genera automáticamente flujos de trabajo de GitHub Actions (.github/workflows/deploy.yaml y .github/workflows/validate.yaml, fácilmente adaptables a GitLab CI o Cloud Build). Para los flujos de trabajo de producción, confirmar tu código y enviar o combinar cambios en la rama main de tu repositorio de GitHub activa automáticamente la acción de GitHub para validar e implementar la canalización en tu entorno de producción de Managed Airflow.

Resumen de la sección: Configuraste la conexión del planificador de Airflow, implementaste tu canalización analítica de extremo a extremo en Managed Airflow y supervisaste una ejecución en vivo, verificando el sistema desde los registros sin procesar hasta las predicciones finales de Cloud Spanner.

9. Limpia

Para evitar que se apliquen cargos continuos a tu proyecto de Google Cloud por los recursos que usaste en este codelab, desmantela el entorno con la secuencia de comandos automatizada.

¡No te saltes el desmontaje! Managed Airflow y Cloud Spanner generan cargos continuos mientras se aprovisionan. Si omites la limpieza, ejecútala más tarde o borra todo el proyecto de Google Cloud que creaste para este lab.

1. En el panel Terminal (o en Cloud Shell), navega al directorio de secuencias de comandos y ejecuta lo siguiente:

cd ~/devrel-demos/codelabs/agentic-data-labs/data-science/scripts
chmod +x teardown.sh
./teardown.sh

2. La secuencia de comandos mostrará todos los recursos que planea borrar y solicitará confirmación:
   • Managed Airflow Environment (cymbal-airflow)
   • Instancia de Cloud Spanner (cymbal-fraud)
   • Conjunto de datos de BigQuery (transactions_dataset_evals)
   • Buckets de Cloud Storage (gs://${PROJECT_ID}-fin-clearing-raw y gs://${PROJECT_ID}-models)
   • Cuenta de servicio del trabajador (composer-worker-sa)
3. Escribe y para confirmar. La secuencia de comandos de desmontaje quitará todos los servicios de GCP aprovisionados y limpiará los archivos locales.

10. ¡Felicitaciones!

Creaste una canalización de detección de fraude de extremo a extremo que abarca Cloud Storage, BigQuery, Managed Service para Apache Spark (Spark Serverless), dbt, Cloud Spanner y Managed Service para Apache Airflow, y programaste en parejas con el kit de agentes de datos de Google Cloud dentro del IDE de Antigravity.

Qué lograste

1. 📥 Registros de transacciones sin procesar transferidos a una tabla de BigQuery con Managed Service para Apache Spark y el Data Agent Kit
2. 🧹 Datos desduplicados y normalizados creando un proyecto de dbt con pruebas de calidad de los datos
3. 🤖 Entrenó un modelo de bosque aleatorio distribuido con RandomForestClassifier y exportó el modelo entrenado a Cloud Storage.
4. ⚡ Ejecutó la inferencia por lotes en las transacciones entrantes y enrutó los registros de alto riesgo a Cloud Spanner para su revisión de auditoría.
5. 🔄 Organizó, implementó y supervisó el flujo de trabajo como un DAG de Airflow programado con Managed Service para Apache Airflow y las herramientas visuales de administración de DAG del IDE.

Conceptos clave

Concepto	Qué aprendiste
Data Agent Kit	Programación en parejas dentro del IDE con lenguaje natural para generar notebooks de PySpark, configurar modelos de dbt y definir DAGs de Airflow
BigQuery	Almacenamiento tabular escalable para SQL analítico, transformaciones de dbt y entrenamiento de AA
Spark sin servidores	Ejecución sin servidores para la carga de datos de PySpark distribuida y el entrenamiento de AA con Random Forest
Cloud Spanner Connector	Escribir predicciones de inferencia de Spark por lotes directamente en las colas de revisión de la base de datos operativa
Declaraciones de DAG en YAML	Definiciones de canalizaciones declarativas renderizadas como gráficos visuales interactivos de Airflow en el IDE
Administración visual de DAG	Inspección de dependencias de canalizaciones, implementación en Managed Airflow y supervisión del historial de ejecución de tareas en vivo dentro del IDE

Próximos pasos

• Explora la documentación del kit de Google Cloud Data Agent
• Obtén más información sobre Managed Service para Apache Spark
• Obtén más información sobre Managed Service para Apache Airflow
• Crea tus propias canalizaciones de varios servicios con el IDE de Antigravity