Pipeline de detecção de fraude com o Data Agent Kit e o IDE do Antigravity

1. Introdução

Imagine que você é um cientista de dados na Cymbal Financial, uma empresa de processamento de pagamentos de alto volume. Houve uma onda de atrasos nos pagamentos, e a equipe de compliance suspeita de fraude coordenada. Você precisa criar um pipeline para ingerir registros brutos de transações da câmara de compensação, limpar os dados, treinar um modelo de machine learning, executar inferência em lote e enviar transações de alto risco para uma fila de revisão do Cloud Spanner para auditoria manual.

Normalmente, isso exige dias de escrita de código de configuração repetitivo (notebooks do Spark, configurações do dbt, scripts de treinamento, DAGs do Airflow) e troca constante de contexto entre interfaces de console e editores.

Neste codelab, você vai programar em dupla com um agente usando o Google Cloud Data Agent Kit (DAK) no IDE do Antigravity. Usando linguagem natural de conversação, o agente vai ajudar você a gerar notebooks do Spark, compilar um projeto do dbt, criar um loop de inferência e orquestrar o fluxo de trabalho usando o Serviço Gerenciado para Apache Airflow.

Atividades deste laboratório

• Ingerir registros da clearinghouse do Cloud Storage usando o Serviço gerenciado para Apache Spark (Spark sem servidor) em uma tabela do BigQuery.
• Remova duplicidades e normalize transações usando o dbt para estabelecer camadas de dados limpas (brutos, de teste e enriquecidos).
• Treine um modelo de classificação de floresta aleatória distribuída (RandomForestClassifier) no Spark sem servidor.
• Execute a inferência em lote em novas transações e grave alertas de alto risco diretamente no Cloud Spanner.
• Orquestre, configure visualmente e implante todo o pipeline usando o Serviço Gerenciado para Apache Airflow e o monitoramento interativo de DAGs no ambiente de desenvolvimento integrado.

O que é necessário

• Um navegador da web, como o Chrome
• Um projeto na nuvem do Google Cloud com o faturamento ativado. Recomendamos usar um projeto novo e dedicado para laboratórios práticos.
• Conhecimento básico de SQL, Python e PySpark.
• IDE do Antigravity com uma assinatura do Google AI Pro (recomendado)

Os recursos criados neste codelab custam menos de US $5. Siga as instruções de Limpeza no final do laboratório para excluir os recursos provisionados.

Escolha sua ferramenta:este laboratório usa a IDE do Antigravity. A extensão funciona em todas as ramificações do VS Code. Em breve, vamos lançar versões alternativas para o VS Code, além de fluxos de trabalho baseados em plug-ins para o Antigravity 2.0, a CLI do Antigravity e o Claude Code.

2. configuração do ambiente

Para começar o laboratório, execute um script de inicialização. Esse script ativa automaticamente as APIs do GCP necessárias, cria um bucket do Cloud Storage para ingestão, gera conjuntos de dados de transação e diretório simulados, carrega diretórios de referência no BigQuery e inicia o provisionamento em segundo plano do Cloud Spanner e do Serviço Gerenciado para Apache Airflow (antes conhecido como Cloud Composer).

Escolher ou criar um projeto

Escolha um projeto atual ou crie um novo no console do Google Cloud.

Verificar o faturamento

Verifique se a cobrança está ativada para o seu projeto do Google Cloud. Para saber mais sobre como fazer isso, siga este guia.

Execute o script de configuração

Você vai usar o Google Cloud Shell (ou o shell local configurado com a CLI do Google Cloud) para iniciar a configuração do ambiente.

1. Abra o Console do Google Cloud.
2. Clique em Ativar o Cloud Shell na barra de ferramentas no canto superior direito.

3. No terminal do Cloud Shell, configure seu projeto ativo:

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

4. Clone o repositório do codelab e navegue até a pasta "scripts":

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. Execute o script de configuração de bootstrap para implantar todos os recursos em us-central1:

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

O que o setup.sh faz:

• Ativa APIs:ativa as APIs BigQuery, Dataproc, Spanner, Airflow Gerenciado, Cloud Storage e de infraestrutura de suporte.
• Cria buckets do Cloud Storage:provisiona buckets para registros de transações brutas e artefatos de modelos de ML.
• Prepara os dados iniciais:copia os registros de transações brutas para o Cloud Storage e carrega os dados de referência no BigQuery.
• Configura o Spark sem servidor:cria o modelo de sessão de ambiente de execução com o conector do Cloud Spanner.
• Provisiona serviços em segundo plano:gera workers em segundo plano para criar o Cloud Spanner e o Airflow Gerenciado.
• Gera .env:salva a configuração do ambiente para o codelab e os scripts de encerramento.

Os recursos criados pelo script de configuração são excluídos na etapa final de limpeza com ./teardown.sh.

6. Quando o script terminar, você vai ver um resumo indicando que o conjunto de dados do BigQuery e o bucket do Cloud Storage estão prontos. Em segundo plano, o Cloud Spanner (leva cerca de 2 minutos) e o Airflow gerenciado (leva cerca de 20 minutos) continuam o provisionamento. É possível monitorar o progresso a qualquer momento executando:

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

Ciclo de desenvolvimento otimizado:como o Airflow Gerenciado leva cerca de 20 minutos para ser provisionado, o script de configuração inicia o processo em segundo plano imediatamente. Isso permite que você trabalhe nas páginas 3 (ingestão), 4 (transformações do dbt), 5 (treinamento de ML) e 6 (inferência em lote) enquanto o Airflow está sendo executado. Quando você chegar à página 7 (Orquestração), seu ambiente estará totalmente pronto.

Abra o IDE do Antigravity

1. Faça o download e instale a IDE do Antigravity na página de download do Google Antigravity.
2. Inicie o IDE do Antigravity.
3. Crie uma pasta vazia na sua máquina local (por exemplo, agentic-data-labs) e abra-a no ambiente de desenvolvimento integrado escolhendo Abrir pasta. Ele vai funcionar como seu espaço de trabalho local para o codelab.

Instalar a extensão do Data Agent Kit

A extensão Google Cloud Data Agent Kit oferece integração avançada com os serviços de dados do Google Cloud diretamente no seu editor. Assim, é possível interagir com o BigQuery, o Cloud SQL, o Cloud Storage e muito mais sem mudar de contexto.

1. No IDE Antigravity, clique no ícone Extensões na barra de atividades, no lado esquerdo da tela (parece quatro quadrados).
2. Na barra de pesquisa na parte de cima do painel "Extensões", digite Google Cloud Data Agent Kit.
3. Localize a extensão chamada Google Cloud Data Agent Kit publicada por googlecloudtools
4. Clique no botão Install.
5. Uma mensagem pode aparecer perguntando: "Você confia no editor 'googlecloudtools' e nas extensões dele?". Clique em Confiar em editores e instalar para continuar.

Depois de instalado, um novo ícone do Google Cloud Data Agent Kit vai aparecer na barra de atividades, no canto esquerdo do IDE do Antigravity.

6. Uma página de integração intitulada "Bem-vindo ao Google Cloud Data Agent Kit" será aberta automaticamente. Se você não estiver conectado à sua conta do Cloud, siga as instruções para permitir o acesso.
7. Na seção Resumo da configuração, localize o campo do projeto. Clique no menu suspenso e selecione seu projeto do Google Cloud. Defina sua região como us-central1. Em seguida, selecione Configurar servidores MCP.

8. Selecione Configurar servidores MCP. No painel Configuração do MCP, ative os seguintes servidores MCP remotos:
   • BigQuery
   • Spanner
   • Notebooks

Em seguida, clique em Começar.

Já instalou ou a página de boas-vindas não abriu? Você pode abrir as configurações de extensão manualmente a qualquer momento. Clique no ícone Kit do agente de dados do Google Cloud na Barra de atividades, no canto esquerdo, e selecione Configurações para ativar ou desativar sua configuração.

Observação sobre habilidades/MCPs ativos:se você tiver outras habilidades personalizadas ou servidores MCP ativos no seu ambiente de desenvolvimento integrado, eles poderão interferir no fluxo esperado deste laboratório. Se o agente se comportar de maneira inesperada, desative-o para este laboratório.

Conhecer opções de configuração

Quando a configuração for concluída, você vai acessar a página "Começar a usar o Google Cloud Data Agent Kit".

1. Em "Configuração", clique em Começar.
2. O painel Configuração do Data Agent Kit será aberto. Confira as guias:
   • Projeto e região:verifique o ID do projeto selecionado e confirme se o script de configuração ativou todas as APIs necessárias (Compute Engine, Cloud Storage, BigQuery, Spanner etc.).
   • BigQuery:configure o local padrão para suas consultas do BigQuery. Use a região us-central1.
   • Configurar servidores MCP:veja os servidores MCP (BigQuery, Notebooks, Spanner etc.) ativados que permitem que agentes de IA interajam com seus dados de forma segura.
   • Habilidades:conheça as habilidades pré-criadas que oferecem aos agentes recursos especializados para tarefas complexas de dados.

Resumo da seção:você executou o script de bootstrap para criar recursos do GCS e do BigQuery enquanto o Spanner e o Airflow eram criados em segundo plano. Em seguida, você abriu o projeto no IDE do Antigravity e ativou a extensão do Google Cloud Data Agent Kit. Agora você já pode escrever seu primeiro notebook.

3. Ingerir registros brutos usando o Spark sem servidor

Nesta seção, você vai ingerir registros de transações JSON brutos no data lake. O Serviço gerenciado para Apache Spark (Spark sem servidor) se conecta diretamente ao armazenamento nativo do BigQuery. Você vai usar o conector padrão do BigQuery para gerenciar dados tabulares e ativar consultas e análises diretas.

Conhecer o ambiente de execução do Spark sem servidor pré-configurado

Antes de executar o código do Spark, inspecione o modelo do ambiente de execução sem servidor pré-configurado pelo script de configuração. Esse modelo define o back-end do ambiente de execução de destino e agrupa as dependências necessárias do conector.

1. Na barra de atividades do IDE, abra o painel Google Cloud Data Agent Kit.
2. Expanda o menu suspenso Apache Spark e depois Serverless.
3. Clique com o botão direito do mouse em fraud-pipeline-runtime e selecione Perfil para abrir a visualização de configuração no editor.
4. Na guia Perfil, role a tela para baixo e expanda Propriedades para inspecionar as dependências personalizadas anexadas ao ambiente:
   • spark.jars: contém gs://spark-lib/spanner/spark-3.5-spanner-1.4.0.jar, que usa o conector do Spark Spanner para permitir que os jobs do Spark gravem resultados de inferência diretamente no Cloud Spanner mais tarde no laboratório. Observação: o Dataproc sem servidor inclui o conector do Spark BigQuery do Google Cloud por padrão, sem exigir configuração adicional de jar para ler e gravar tabelas do BigQuery.

5. Observe a guia Sessões interativas à esquerda. No momento, ela está vazia porque você ainda não executou nenhum código. Assim que você executar o notebook na próxima etapa, uma sessão de computação sem servidor ativa será provisionada dinamicamente e vai aparecer aqui.

Ingerir dados usando o Data Agent Kit

Em vez de configurar manualmente uma sessão do Spark ou escrever scripts de carregamento do PySpark do zero, você vai programar em dupla com um agente usando o Data Agent Kit.

1. Abra o painel Chat do agente clicando no ícone Ativar/desativar agente na barra de ferramentas no canto superior direito.
2. Cole o comando a seguir na conversa. Não se esqueça de substituir ${PROJECT_ID} pelo ID do projeto do 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. Se o agente pedir permissão para executar comandos de verificação em segundo plano (por exemplo, "Permitir a execução deste comando?"), revise o comando proposto e selecione Sim, permitir desta vez ou Sim, e sempre permitir.
4. Quando o agente terminar de gerar o arquivo, clique no botão azul Aceitar tudo (ou no ícone de marca de seleção) na parte de baixo do painel de chat para salvar notebooks/01_ingestion.ipynb no seu espaço de trabalho.

Aprovações de comandos e aceitação de mudanças:como um programador em dupla de IA, o agente pode pedir permissão para executar comandos seguros do terminal (como verificações de sintaxe ou inspeção de dependências) antes de gerar o código. Revise e permita esses comandos quando solicitado e clique em Continuar ou Aceitar tudo no painel de chat sempre que arquivos ou planos de implementação forem apresentados.

Revisar e executar o notebook

1. Abra o notebooks/01_ingestion.ipynb recém-gerado no IDE.
2. Analise o código PySpark da lógica de gravação do conector do BigQuery.
3. Clique em Executar tudo na barra de ferramentas do notebook da IDE.
4. Se esta for a primeira vez que você executa um notebook remoto do Spark, o ambiente de desenvolvimento integrado pode pedir que você instale dependências locais. Se solicitado, clique em Instalar dependências para kernels remotos do Spark e confirme as caixas de diálogo de instalação. Depois, clique em Executar tudo novamente.
5. No menu suspenso Selecionar kernel, escolha Kernels remotos do Spark -> fraud-pipeline-runtime no Spark sem servidor. Dica: se o modelo de tempo de execução pré-configurado não aparecer, clique no ícone de atualização no canto superior direito do menu suspenso do seletor de kernel para recarregar os kernels remotos disponíveis.
6. Confira a barra de status no canto inferior esquerdo do editor. Você vai ver Connecting to kernel: fraud-pipeline-runtime on Serverless Spark.... Como este é o lançamento inicial do back-end do kernel do ambiente de execução do Spark sem servidor, ele vai levar alguns minutos para ser provisionado e inicializado.
7. Depois que o kernel terminar de se conectar, o notebook vai começar a executar todas as células em sequência para processar os registros de transações brutas no conjunto de dados do BigQuery.

Conexão com o Spark Serverless:ao se conectar à sessão de execução do Spark Serverless, você verá Connecting to kernel... na barra de status inferior. Leva cerca de 3 minutos para inicializar os recursos no Google Cloud antes que o notebook se conecte e execute todas as células.

Verificação

Quando a execução for concluída, verifique o catálogo do Data Agent Kit para verificar a criação da tabela:

1. Na barra de atividades do IDE, abra o painel Google Cloud Data Agent Kit.
2. Expanda a seção CATÁLOGO.
3. Expanda o ID do projeto.
4. Expanda BigQuery.
5. Expanda o conjunto de dados transactions_dataset_evals.
6. Clique na tabela raw_transactions para abrir a visualização de detalhes dela no editor principal.
7. Na navegação à esquerda, acesse as guias Dados, Esquema e Detalhes para inspecionar os registros e metadados ingeridos.

Resumo da seção:você usou linguagem natural no chat do agente para gerar uma carga de trabalho completa do Spark sem servidor. Em seguida, você o executou para processar registros JSON não estruturados em uma tabela bruta do BigQuery.

4. Eliminar duplicação e normalizar com dbt

Antes de treinar o modelo de ML, você vai aplicar a qualidade de dados removendo registros de streaming duplicados, isolando registros ruins (como IDs de transação vazios) e unindo dados dimensionais (pagadores e beneficiários). Esse processo exige transformações SQL idempotentes e confiáveis, o que torna o dbt (data build tool) uma ótima opção.

Criar a estrutura do pipeline do dbt

Use o agente para gerar um projeto dbt no conjunto de dados do BigQuery:

1. Volte para o painel Chat do agente.
2. Forneça a seguinte instrução para gerar o projeto do 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. O agente vai apresentar um artefato de Plano de implementação no painel principal do editor. Analise a estrutura de arquivo e a lógica SQL propostas.
4. Clique em Continuar e em Aceitar tudo para permitir que o agente gere os arquivos no seu espaço de trabalho.

5. Quando a geração for concluída, o agente vai mostrar um Tutorial resumindo os novos componentes. Aceite todas as mudanças, se solicitado.

Gerenciamento inteligente de dependências:ao criar projetos de scaffolding, o agente inspeciona proativamente o ambiente do shell em busca das ferramentas de CLI necessárias (como dbt). Se estiverem faltando, o agente poderá oferecer automaticamente a criação de um ambiente virtual e a instalação de dbt-bigquery em segundo plano, solicitando sua aprovação antes de executar os comandos de configuração.

Criar e testar

Embora o agente tenha executado automaticamente dbt compile para garantir que o SQL gerado fosse sintaticamente válido, agora você vai materializar essas visualizações e tabelas no BigQuery e executar os testes de qualidade de dados para verificação local. Observação: mais adiante no laboratório, você vai automatizar essa etapa do dbt como parte de um DAG do Airflow de ponta a ponta.

1. Na barra de atividades à esquerda, clique no ícone Explorer (ou pressione Cmd/Ctrl+Shift+E).
2. Expanda dbt_project -> models para inspecionar os modelos SQL gerados. Clique em enriched_transactions.sql para abrir e revisar a lógica de transformação e de recursos de fraude no editor.
3. No Explorador de Arquivos, clique com o botão direito do mouse na pasta dbt_project e selecione Abrir no terminal integrado. Isso abre automaticamente um painel de terminal definido diretamente no diretório de trabalho dbt_project necessário.
4. Se você ainda não tiver o dbt instalado, crie um ambiente virtual fora do dbt_project/ (na raiz da sua casa ou espaço de trabalho) e instale o adaptador do BigQuery:

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

5. Execute os modelos do dbt e os testes de qualidade de dados associados:

dbt build

6. Observe a saída do terminal. O dbt vai compilar o SQL, materializar as tabelas de preparo e enriquecidas no BigQuery e executar os testes de dados.

Como resolver problemas de builds com o agente:se o dbt build encontrar algum erro (como incompatibilidades de esquema ou relações ausentes), copie e cole a saída de erro diretamente no chat do agente. O agente vai analisar a falha, ajustar os modelos e resolver o problema para você.

7. Quando o build terminar, feche o painel do terminal para liberar espaço na tela para as etapas restantes.

Resumo da seção:você gerou um projeto do dbt com o agente, executou testes de qualidade de dados e transformou os registros brutos em tabelas de preparo e enriquecidas do BigQuery.

5. Treinar um modelo distribuído de detecção de fraudes com floresta aleatória

Com as transações enriquecidas materializadas no BigQuery, você vai criar um modelo de machine learning para classificar eventos fraudulentos. A Floresta aleatória é um método de aprendizado de conjunto adequado para dados de classificação tabular. Executar um RandomForestClassifier no Spark sem servidor distribui o treinamento de modelo entre nós de trabalho sem exigir que você gerencie a infraestrutura.

Nesta etapa, você vai usar o agente para gerar o pipeline de treinamento do Spark ML.

Gerar o notebook de treinamento de ML

1. Abra o painel Chat do agente.
2. Forneça o seguinte comando para projetar a sequência de treinamento do modelo. Não se esqueça de substituir ${PROJECT_ID} pelo ID do seu projeto ativo:

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. Revise o plano ou o código gerado do agente e clique em Continuar / Aceitar tudo para salvar notebooks/02_training.ipynb no seu espaço de trabalho.

Revisar e executar o notebook

1. Abra notebooks/02_training.ipynb no editor.
2. Analise as etapas do pipeline de ML do PySpark para codificação de recursos, montagem de vetores e lógica de classificação de Floresta aleatória.
3. Clique em Executar tudo na barra de ferramentas do notebook da IDE.
4. Quando o seletor suspenso Selecionar kernel for aberto, selecione fraud-pipeline-runtime no Spark sem servidor.

Verificação

Depois que a execução for concluída, confirme se o modelo foi treinado e exportado corretamente:

1. Analise as saídas da célula de avaliação na parte de baixo do notebook para verificar a pontuação da área sob a curva ROC (AUC) informada.
2. Para garantir que os artefatos do modelo foram salvos no GCS, expanda o painel do explorador STORAGE na barra lateral do Data Agent Kit.
3. Localize o bucket que termina em -models (vinculado ao ID do projeto ativo), expanda-o e detalhe para verificar se o diretório fraud_model e os estágios do pipeline existem.

Do notebook à produção:embora os notebooks interativos sejam ótimos para experimentação e desenvolvimento de modelos, seu agente também pode refatorar o código do notebook em scripts PySpark autônomos parametrizáveis para execução em lote recorrente no Dataproc sem servidor.

Resumo da seção:você usou o agente para criar um pipeline de treinamento de ML do PySpark, treinou um modelo de floresta aleatória na sua tabela enriquecida do BigQuery e exportou o modelo para o Cloud Storage.

6. Inferência em lote e gravação do Cloud Spanner

Com um modelo preditivo treinado armazenado no Cloud Storage, você vai executar a inferência em lote em novas transações que passam pelo BigQuery. Transações de alto risco precisam ser encaminhadas para um sistema operacional para que uma equipe de compliance possa revisá-las. O Cloud Spanner oferece um banco de dados transacional escalonável para essa fila de revisão.

Gerar o notebook de inferência em lote

Use o agente para criar um notebook de inferência que conecte o BigQuery, o Cloud Storage e o Cloud Spanner:

1. Abra o painel Chat do agente.
2. Envie o seguinte comando:

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. Aceite o notebook gerado para salvar notebooks/03_inference.ipynb no seu espaço de trabalho.

Por que 50% é alto risco:embora 50% pareça um cara ou coroa, a grande maioria das transações legítimas tem uma pontuação inferior a 5%. Uma pontuação de 50%ou mais representa os 2% principais de transações mais anômalas na rede, filtrando ruídos para que os auditores revisem apenas eventos genuínos de alto risco.

Revisar e executar o notebook

1. Abra o notebooks/03_inference.ipynb recém-gerado no editor.
2. Revise a sequência de inferência do PySpark:
   • Dependências:o modelo de ambiente de execução sem servidor fornece as dependências JAR cloud-spanner necessárias para a execução do Spark.
   • Formatação de dados:o script descarta colunas de vetores complexos do Spark ML (como recursos brutos e probabilidades) antes de gravar para corresponder ao esquema da tabela do Spanner.
   • Conector do Spanner:grava as linhas sinalizadas usando .format("cloud-spanner") para anexar diretamente à fila de revisão.
3. Clique em Executar tudo na barra de ferramentas do notebook da IDE.
4. Quando for solicitado que você selecione um kernel, escolha fraud-pipeline-runtime no Serverless Spark.

Verificação

Quando o notebook de inferência terminar o processamento, você poderá consultar seu banco de dados operacional do Spanner diretamente no ambiente de desenvolvimento integrado:

1. Na barra de atividades do IDE, abra o painel Google Cloud Data Agent Kit.
2. Expanda a seção CATÁLOGO.
3. Expanda o ID do projeto e, em seguida, Spanner.
4. Navegue até cymbal-fraud -> fraud-db -> Tables -> SparkEvalFraudReviewQueue.
5. Clique com o botão direito do mouse na tabela e selecione Consultar tabela. Depois, execute a consulta:

SELECT *
FROM `SparkEvalFraudReviewQueue`
LIMIT 100;

6. No painel Resultados da consulta abaixo, você vai encontrar linhas recém-inseridas que representam transações de alto risco sinalizadas para revisão manual.

Resumo da seção:você usou o agente para criar um notebook de inferência em lote, classificou registros não rotulados do BigQuery com seu modelo treinado e gravou transações de alto risco diretamente no Cloud Spanner.

7. Estruturar e orquestrar com o Airflow gerenciado

No momento, seu pipeline consiste em etapas discretas: um notebook de ingestão, um projeto de transformação do dbt e um notebook de inferência em lote. Para tornar isso pronto para produção, você vai juntá-los em um gráfico de dependência programado.

O Serviço gerenciado para Apache Airflow (antigo Cloud Composer) oferece um mecanismo de orquestração gerenciado para esse fluxo de trabalho. O Data Agent Kit inclui um recurso de Orquestração de Pipelines que traduz definições declarativas de pipelines YAML diretamente em DAGs do Airflow.

Definir o pipeline

Use o agente para gerar a configuração do pipeline de orquestração:

1. No Chat do agente, insira o seguinte comando (lembre-se de substituir ${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.

Revisar a configuração do DAG

O orquestrador do Data Agent Kit usa configurações YAML declarativas para definir e implantar pipelines no Apache Airflow, permitindo que as definições sejam controladas por versões e implantadas via CI/CD.

No painel do IDE Explorer, revise os dois arquivos de pipeline que o agente gerou na raiz do seu espaço de trabalho:

1. deployment.yaml: abra este arquivo. Ele serve como seu registro de ambiente. Ele mapeia o pipeline lógico dev para o ambiente cymbal-airflow, define a região de execução (us-central1) e define o bucket artifact_storage em que os DAGs e as dependências compilados são armazenados.
2. fraud_analysis_pipeline.yaml: abra este arquivo. Isso define o gráfico de execução. Ele especifica a programação do gatilho (interval: '0 0 * * *') e sequencia as três etapas no bloco actions:
   • Uma ação de ingestão notebook para 01_ingestion.ipynb executado no Dataproc sem servidor.
   • Uma ação de transformação pipeline que segmenta o diretório dbt_project, com uma dependência dependsOn apontando para a etapa de ingestão.
   • Uma ação de inferência notebook para 03_inference.ipynb com uma dependência dependsOn que aponta para a etapa do dbt, agrupando a propriedade JAR do Spanner.
3. O agente também vai resumir esses artefatos gerados em uma guia Tutorial no painel do editor, descrevendo as configurações e validações realizadas.

Configuração interativa de DAGs

O Data Agent Kit renderiza a configuração do pipeline como um gráfico visual interativo para inspecionar e editar propriedades de DAG do Airflow.

1. Na barra de atividades do IDE, abra o painel Google Cloud Data Agent Kit.
2. Em DATA ENGINEERING, expanda Orchestration Pipelines.
3. Clique em fraud_analysis_pipeline.yaml para abrir o canvas do DAG visual no editor principal.

4. Clique no nó Schedule trigger na parte de cima. Um menu suspenso de configuração é aberto à direita, mostrando a string Cron analisada (0 0 * * *) e permitindo ajustar parâmetros como backfill e catchup.
5. Clique em qualquer nó de tarefa do notebook, como a etapa de ingestão ou inferência. O painel flutuante é atualizado para mostrar os mapeamentos de execução e as propriedades do conector específicos do Dataproc sem servidor.
6. Observe o hiperlink do nome do arquivo do notebook (como 01_ingestion.ipynb) dentro do bloco do nó. Ao clicar nele, o notebook é aberto diretamente no editor.
7. Na barra lateral esquerda, abaixo de "Orquestração de Pipelines", clique em Deployment configuration. Essa visualização mostra o cluster de ambiente dev de destino e os artefatos do bucket do GCS de saída.

Resumo da seção:você gerou uma configuração de pipeline de orquestração com o agente, definindo dependências entre tarefas de ingestão, dbt e inferência em uma tela visual interativa.

8. Implantar, executar e monitorar

Com o DAG definido localmente, você vai se conectar ao ambiente do Airflow gerenciado provisionado durante a configuração e implantar o pipeline.

Configurar o Serviço gerenciado para Apache Airflow

Antes da implantação, configure a conexão do Scheduler nas configurações do Data Agent Kit para que a extensão tenha como destino seu ambiente do Airflow Gerenciado:

1. Na barra de atividades do IDE, abra o painel Google Cloud Data Agent Kit.
2. Em SETTINGS, clique em Configurações.
3. Selecione Programador no menu à esquerda.
4. Defina as configurações:
   • ID do projeto: selecione o ID do projeto ativo.
   • Região: selecione us-central1.
   • Ambiente: selecione cymbal-airflow.
5. Clique em Salvar.

Implante o DAG

Otimização de artefatos de implantação:antes de implantar, verifique se pastas temporárias como .venv, target/ ou logs/ não estão aninhadas no diretório dbt_project/. Fazer upload de ambientes virtuais grandes com milhares de arquivos pode fazer com que o empacotamento fique travado e desacelere a análise de DAGs do Airflow. Se estiverem presentes, remova-os (rm -rf dbt_project/.venv dbt_project/target dbt_project/logs) ou adicione-os a um arquivo .airflowignore.

Agora, implante o pipeline configurado diretamente no ambiente do Airflow Gerenciado na tela visual:

1. Na barra lateral Kit do agente de dados do Google Cloud, expanda DATA ENGINEERING > Orchestration Pipelines e clique em fraud_analysis_pipeline.yaml para abrir o canvas do DAG visual.
2. No canto superior direito da barra de ferramentas da tela, clique no botão azul Executar pipeline.
3. No seletor do menu suspenso de ambiente, selecione dev.
4. Observe a notificação de progresso na área de status da parte de baixo (Running pipeline: Building pipeline locally...). A extensão vai compilar automaticamente seu DAG, empacotar o notebook e os recursos do dbt e fazer upload deles para o bucket do GCS do seu ambiente do Airflow Gerenciado. Isso leva de 3 a 4 minutos para ser concluído.

Monitorar a execução

Quando a compilação local for concluída e a notificação pop-up confirmar Triggered a new run for pipeline... successfully, monitore a execução ativa:

1. Na barra lateral Google Cloud Data Agent Kit, expanda DATA ENGINEERING > Orchestration Pipelines.
2. Clique em Gerenciamento de pipelines.
3. Na tabela "Gerenciamento de pipelines", clique em fraud_analysis_pipeline para abrir o histórico de execução.

4. Na visualização Histórico de execução, selecione a execução ativa no calendário.
5. À medida que a execução avança em cada tarefa do pipeline (ingestão, transformação do dbt e inferência), os indicadores de status são atualizados e as durações das tarefas são preenchidas. Clique em qualquer tarefa para inspecionar a saída de execução em tempo real e os registros do DAG do Airflow.

Implantações de produção usando CI/CD e ações do GitHub:além das implantações locais, o Data Agent Kit gera automaticamente fluxos de trabalho de ações do GitHub (.github/workflows/deploy.yaml e .github/workflows/validate.yaml, facilmente adaptáveis ao GitLab CI ou ao Cloud Build). Para fluxos de trabalho de produção, fazer commit do código e enviar ou mesclar mudanças na ramificação main do repositório do GitHub aciona automaticamente a ação do GitHub para validar e implantar o pipeline no ambiente de Airflow Gerenciado de produção.

Resumo da seção:você configurou a conexão do escalonador do Airflow, implantou seu pipeline analítico completo no Airflow Gerenciado e monitorou uma execução em tempo real, verificando o sistema desde os registros brutos até as previsões finais do Cloud Spanner.

9. Limpar

Para evitar cobranças contínuas no seu projeto na nuvem do Google Cloud pelos recursos usados neste codelab, desative o ambiente usando o script automatizado.

Não pule a desmontagem! O Airflow Gerenciado e o Cloud Spanner geram cobranças contínuas enquanto são provisionados. Se você pular a desmontagem, execute-a mais tarde ou exclua todo o projeto na nuvem do Google Cloud criado para este laboratório.

1. No painel Terminal (ou no Cloud Shell), navegue até o diretório de scripts e execute:

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

2. O script vai listar todos os recursos que planeja excluir e pedir confirmação:
   • Ambiente do Airflow gerenciado (cymbal-airflow)
   • Instância do Cloud Spanner (cymbal-fraud)
   • Conjunto de dados do BigQuery (transactions_dataset_evals)
   • Buckets do Cloud Storage (gs://${PROJECT_ID}-fin-clearing-raw e gs://${PROJECT_ID}-models)
   • Conta de serviço do worker (composer-worker-sa)
3. Digite y para confirmar. O script de encerramento vai remover todos os serviços provisionados do GCP e limpar os arquivos locais.

10. Parabéns!

Você criou um pipeline de detecção de fraude de ponta a ponta que abrange o Cloud Storage, o BigQuery, o Serviço Gerenciado para Apache Spark (Spark sem servidor), o dbt, o Cloud Spanner e o Serviço Gerenciado para Apache Airflow, com programação em dupla usando o Google Cloud Data Agent Kit no IDE do Antigravity.

O que você realizou

1. 📥 Ingerimos registros de transações brutas em uma tabela do BigQuery usando o Serviço Gerenciado para Apache Spark e o Data Agent Kit.
2. 🧹 Dados duplicados e normalizados criando um projeto do dbt com testes de qualidade de dados.
3. 🤖 Treinei um modelo de floresta aleatória distribuída usando RandomForestClassifier e exportei o modelo treinado para o Cloud Storage.
4. ⚡ Executamos a inferência em lote nas transações recebidas e encaminhamos registros de alto risco para o Cloud Spanner para revisão de auditoria.
5. 🔄 Orquestrou, implantou e monitorou o fluxo de trabalho como um DAG do Airflow programado usando o Serviço Gerenciado para Apache Airflow e as ferramentas visuais de gerenciamento de DAG do ambiente de desenvolvimento integrado.

Principais conceitos

Conceito	O que você aprendeu
Data Agent Kit	Programação em dupla no IDE usando linguagem natural para gerar notebooks PySpark, configurar modelos dbt e definir DAGs do Airflow
BigQuery	Armazenamento tabular escalonável para SQL analítico, transformações do dbt e treinamento de ML
Spark sem servidor	Execução sem servidor para carregamento de dados distribuídos do PySpark e treinamento de ML de floresta aleatória
Conector do Cloud Spanner	Gravar previsões de inferência em lote do Spark diretamente nas filas de revisão do banco de dados operacional
Declarações de DAG em YAML	Definições declarativas de pipeline renderizadas como gráficos visuais interativos do Airflow na IDE
Gerenciamento visual de DAGs	Inspecionar dependências de pipeline, implantar no Airflow gerenciado e monitorar o histórico de execução de tarefas ativas no ambiente de desenvolvimento integrado

Próximas etapas

• Confira a documentação do Google Cloud Data Agent Kit.
• Saiba mais sobre o Serviço Gerenciado para Apache Spark
• Saiba mais sobre o Serviço gerenciado para Apache Airflow.
• Crie seus próprios pipelines de vários serviços usando o IDE do Antigravity