Конвейер обнаружения мошенничества с использованием Data Agent Kit и Antigravity IDE

1. Введение

Представьте, что вы — специалист по анализу данных в компании Cymbal Financial , занимающейся обработкой больших объемов платежей. Произошла волна задержек в расчетах, и команда по соблюдению нормативных требований подозревает скоординированное мошенничество. Вам необходимо создать конвейер для обработки необработанных журналов транзакций клиринговой палаты, очистки данных, обучения модели машинного обучения, выполнения пакетного вывода и отправки транзакций высокого риска в очередь проверки Cloud Spanner для ручного аудита.

Обычно это требует нескольких дней написания повторяющегося кода настройки (блокноты Spark, конфигурации dbt, скрипты обучения, DAG-графы Airflow) и постоянного переключения между консольными интерфейсами и редакторами.

В этом практическом занятии вы будете работать в паре с агентом, используя Google Cloud Data Agent Kit (DAK) внутри среды разработки Antigravity IDE . Используя разговорный естественный язык, агент поможет вам создавать блокноты Spark, компилировать проект dbt, создавать цикл вывода и управлять рабочим процессом с помощью управляемого сервиса Apache Airflow .

Что вы будете делать

• Загрузка журналов из облачного хранилища с использованием управляемого сервиса для Apache Spark (Spark Serverless) в таблицу BigQuery .
• С помощью dbt можно выполнить дедупликацию и нормализацию транзакций для создания чистых слоев данных (исходные, промежуточные, обогащенные).
• Обучите распределенную модель классификации Random Forest ( RandomForestClassifier ) ​​на Spark Serverless.
• Запускайте пакетный анализ новых транзакций и записывайте оповещения о высоком риске непосредственно в Cloud Spanner .
• Организуйте, визуально настройте и разверните весь конвейер с помощью управляемого сервиса для Apache Airflow и интерактивного мониторинга DAG внутри IDE.

Что вам понадобится

• Веб-браузер, например Chrome.
• Проект Google Cloud с включенной оплатой (для практических занятий рекомендуется использовать новый, выделенный проект).
• Базовые знания SQL, Python и PySpark.
• Antigravity IDE с подпиской Google AI Pro (рекомендуется)

Стоимость ресурсов, созданных в этом практическом задании, должна быть менее 5 долларов. Обязательно следуйте инструкциям по очистке в конце задания, чтобы удалить выделенные ресурсы.

Выберите инструмент: В этой лабораторной работе используется IDE Antigravity (расширение работает во всех версиях VS Code). Альтернативные версии для VS Code, а также рабочие процессы на основе плагинов для Antigravity 2.0, Antigravity CLI и Claude Code появятся в ближайшее время!

2. Настройка среды

Для начала лабораторной работы вам потребуется запустить скрипт начальной загрузки. Этот скрипт автоматически включает необходимые API GCP, создает сегмент Cloud Storage для приема данных, генерирует фиктивные наборы данных транзакций и каталогов, загружает справочные каталоги в BigQuery и запускает фоновое выделение ресурсов Cloud Spanner и управляемого сервиса для Apache Airflow (ранее известного как Cloud Composer).

Выберите или создайте проект

Выберите существующий проект или создайте новый проект в консоли Google Cloud.

Подтвердите выставление счетов.

Убедитесь, что для вашего проекта Google Cloud включена функция выставления счетов. Подробнее о том, как это сделать, вы можете узнать, следуя этому руководству .

Запустите скрипт установки.

Для запуска настройки среды вам потребуется использовать Google Cloud Shell (или локальную оболочку, настроенную с помощью Google Cloud CLI).

1. Откройте консоль Google Cloud .
2. Нажмите кнопку «Активировать Cloud Shell» на панели инструментов в правом верхнем углу.

3. В терминале Cloud Shell настройте свой активный проект:

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

4. Клонируйте репозиторий codelab и перейдите в папку 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. Запустите скрипт начальной загрузки, чтобы развернуть все ресурсы в us-central1 :

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

Что делает setup.sh :

• Включает API: активирует BigQuery, Dataproc, Spanner, Managed Airflow, Cloud Storage и вспомогательные инфраструктурные API.
• Создает сегменты облачного хранилища: выделяет сегменты для необработанных журналов транзакций и артефактов моделей машинного обучения.
• Подготовка исходных данных: копирование необработанных журналов транзакций в облачное хранилище и загрузка справочных данных в BigQuery.
• Настройка Spark Serverless: создание шаблона сессии во время выполнения с помощью коннектора Cloud Spanner.
• Запуск фоновых служб: создание фоновых рабочих процессов для сборки Cloud Spanner и Managed Airflow.
• Создает файл .env : сохраняет конфигурацию среды для скриптов выполнения практического задания и завершения работы.

Ресурсы, созданные скриптом установки, удаляются на заключительном этапе очистки с помощью ./teardown.sh .

6. После завершения выполнения скрипта вы увидите сводную информацию о готовности вашего набора данных BigQuery и хранилища Cloud Storage. В фоновом режиме Cloud Spanner (занимает около 2 минут) и Managed Airflow (занимает около 20 минут) продолжат подготовку ресурсов. Вы можете отслеживать ход выполнения в любое время, выполнив команду:

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

Оптимизированный цикл разработки: поскольку развертывание Managed Airflow занимает около 20 минут, скрипт настройки запускает его в фоновом режиме немедленно. Это позволяет вам работать на страницах 3 (Загрузка данных), 4 (Преобразования dbt), 5 (Обучение машинного обучения) и 6 (Пакетный вывод), пока Airflow находится в процессе развертывания. К моменту достижения страницы 7 (Оркестрация) ваша среда будет полностью готова.

Откройте Antigravity IDE

1. Загрузите и установите Antigravity IDE со страницы загрузки Google Antigravity .
2. Запустите среду разработки Antigravity IDE .
3. Создайте на локальном компьютере новую пустую папку (например, с именем agentic-data-labs ) и откройте её в IDE, выбрав «Открыть папку» . Она будет служить вашей локальной рабочей областью для этого практического задания.

Установите расширение Data Agent Kit.

Расширение Google Cloud Data Agent Kit обеспечивает глубокую интеграцию с сервисами данных Google Cloud непосредственно в вашем редакторе, позволяя взаимодействовать с BigQuery, Cloud SQL, Cloud Storage и другими сервисами без переключения контекста.

1. В среде разработки Antigravity IDE щелкните значок «Расширения» на панели активности в левой части экрана (он выглядит как четыре квадрата).
2. В строке поиска в верхней части панели расширений введите Google Cloud Data Agent Kit .
3. Найдите расширение под названием Google Cloud Data Agent Kit , опубликованное на сайте googlecloudtools
4. Нажмите кнопку «Установить» .
5. Возможно, появится запрос: «Доверяете ли вы издателю 'googlecloudtools' и его расширениям?». Нажмите «Доверять издателям и установить» , чтобы продолжить.

После установки в левой части панели активности среды разработки Antigravity IDE появится новый значок Google Cloud Data Agent Kit .

6. Автоматически должна открыться страница регистрации под названием «Добро пожаловать в Google Cloud Data Agent Kit». Если вы не вошли в свою учетную запись Cloud, следуйте инструкциям, чтобы разрешить доступ.
7. В разделе «Сводка конфигурации» найдите поле «Проект». Щелкните раскрывающийся список и выберите свой проект Google Cloud. Установите регион как us-central1 . Затем выберите «Настроить серверы MCP» .

8. Выберите «Настроить серверы MCP» . В панели «Конфигурация MCP» убедитесь, что включены следующие удаленные серверы MCP:
   • BigQuery
   • Гаечный ключ
   • Тетради

Затем нажмите «Начать» .

Расширение уже установлено или приветственная страница не открылась? Вы можете в любой момент открыть настройки расширения вручную. Щелкните значок Google Cloud Data Agent Kit на панели действий в левом верхнем углу, затем выберите «Настройки» , чтобы изменить конфигурацию.

Примечание об активных навыках/MCP: Если в вашей IDE активны другие пользовательские навыки или серверы MCP, они могут помешать нормальному выполнению этой лабораторной работы. Если агент ведет себя непредсказуемо, рассмотрите возможность его отключения для этой лабораторной работы.

Изучите параметры конфигурации.

После завершения настройки вы попадете на страницу "Начало работы с Google Cloud Data Agent Kit".

1. В разделе «Настройка и конфигурация» нажмите «Начать» .
2. Это откроет панель настройки Data Agent Kit . Изучите вкладки:
   • Проект и регион: Проверьте выбранный идентификатор проекта и убедитесь, что скрипт настройки включил все необходимые API (Compute Engine, Cloud Storage, BigQuery, Spanner и т. д.).
   • BigQuery: Настройте местоположение по умолчанию для ваших запросов BigQuery. Используйте регион us-central1 .
   • Настройка серверов MCP: Просмотрите список включенных серверов MCP (BigQuery, Notebooks, Spanner и т. д.), которые позволяют агентам ИИ безопасно взаимодействовать с вашими данными.
   • Навыки: Изучите встроенные навыки , которые предоставляют агентам специализированные возможности для решения сложных задач обработки данных.

Краткое содержание раздела: Вы запустили скрипт начальной загрузки для создания ресурсов GCS и BigQuery, в то время как Spanner и Airflow выполняли сборку в фоновом режиме. Затем вы открыли проект в IDE Antigravity и активировали расширение Google Cloud Data Agent Kit. Теперь вы готовы написать свой первый блокнот.

3. Загрузка необработанных логов с помощью Spark Serverless.

В этом разделе вы будете загружать необработанные JSON-журналы транзакций в озеро данных. Управляемая служба для Apache Spark (Spark Serverless) напрямую подключается к собственному хранилищу BigQuery . Вы будете использовать стандартный коннектор BigQuery для управления табличными данными и обеспечения прямого выполнения запросов и аналитики.

Изучите предварительно настроенную среду выполнения Spark Serverless.

Перед выполнением кода Spark проверьте шаблон среды выполнения Serverless Runtime, предварительно настроенный скриптом установки. Этот шаблон определяет целевую среду выполнения и включает необходимые зависимости коннектора.

1. В панели действий IDE откройте панель Google Cloud Data Agent Kit .
2. Разверните выпадающее меню Apache Spark , затем разверните раздел Serverless .
3. Щелкните правой кнопкой мыши fraud-pipeline-runtime и выберите «Профиль» , чтобы открыть его конфигурационный раздел в редакторе.
4. На вкладке «Профиль» прокрутите вниз и разверните раздел «Свойства» , чтобы просмотреть пользовательские зависимости, подключенные к среде:
   • spark.jars : Содержит gs://spark-lib/spanner/spark-3.5-spanner-1.4.0.jar , который использует коннектор Spark Spanner , позволяющий заданиям Spark записывать результаты вывода непосредственно в Cloud Spanner позже в ходе лабораторной работы. (Примечание: Dataproc Serverless по умолчанию включает коннектор Spark BigQuery от Google Cloud, поэтому для чтения и записи таблиц BigQuery не требуется дополнительная настройка jar-файлов).

5. Обратите внимание на вкладку «Интерактивные сессии» слева. Сейчас она пуста, поскольку вы еще не выполнили никакого кода. Как только вы запустите ноутбук на следующем шаге, динамически будет создана и отобразится активная сессия бессерверных вычислений!

Загрузка данных с помощью Data Agent Kit.

Вместо того чтобы вручную настраивать сессию Spark или писать с нуля скрипты загрузки PySpark, вы будете работать в паре с агентом, используя Data Agent Kit.

1. Откройте панель чата с агентом , нажав на значок «Переключить агента» в верхней правой части панели инструментов.
2. Вставьте следующую подсказку в чат (обязательно замените ${PROJECT_ID} на фактический идентификатор вашего проекта в 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. Если агент запрашивает разрешение на выполнение команд проверки в фоновом режиме (например , "Разрешить выполнение этой команды?" ), просмотрите предложенную команду и выберите "Да, разрешить на этот раз" (или "Да, и всегда разрешать ").
4. Когда агент завершит создание файла, нажмите синюю кнопку «Принять все » (или значок галочки) в нижней части панели чата, чтобы сохранить notebooks/01_ingestion.ipynb в свою рабочую область.

Подтверждение команд и принятие изменений: В качестве программиста-партнера по ИИ агент может запросить разрешение на выполнение безопасных команд терминала (таких как проверка синтаксиса или анализ зависимостей) перед генерацией кода. Просмотрите и разрешите эти команды, когда появится соответствующий запрос, и нажмите «Продолжить» или «Принять все» в панели чата всякий раз, когда отображаются файлы или планы реализации.

Просмотрите и запустите блокнот.

1. Откройте созданный файл notebooks/01_ingestion.ipynb в IDE.
2. Изучите код PySpark, чтобы ознакомиться с логикой записи данных в коннектор BigQuery.
3. В панели инструментов блокнота IDE нажмите кнопку «Запустить все» .
4. Если вы впервые запускаете удаленный блокнот Spark, IDE может предложить вам установить локальные зависимости. В этом случае нажмите «Установить зависимости для удаленных ядер Spark» и подтвердите изменения в диалоговых окнах установки, затем снова нажмите «Запустить все» .
5. В раскрывающемся меню «Выбрать ядро» выберите «Удаленные ядра Spark» -> fraud-pipeline-runtime на Serverless Spark . (Совет: если вы не видите свой предварительно настроенный шаблон среды выполнения в списке, нажмите значок обновления в правом верхнем углу раскрывающегося списка выбора ядра, чтобы перезагрузить доступные удаленные ядра).
6. Посмотрите на строку состояния в левом нижнем углу редактора. Вы увидите Connecting to kernel: fraud-pipeline-runtime on Serverless Spark... . Поскольку это первоначальный запуск бэкенда ядра среды выполнения Spark Serverless, потребуется несколько минут для подготовки и загрузки.
7. После завершения подключения ядра, ноутбук автоматически начнет последовательное выполнение всех ячеек для обработки необработанных журналов транзакций и их переноса в ваш набор данных BigQuery.

Подключение к Spark Serverless: При подключении к сеансу среды выполнения Spark Serverless в нижней строке состояния вы увидите Connecting to kernel... . Инициализация ресурсов в Google Cloud занимает около 3 минут, прежде чем ноутбук подключится и выполнит все ячейки.

Проверка

После завершения выполнения проверьте каталог Data Agent Kit, чтобы убедиться в создании таблицы:

1. В панели действий IDE откройте панель Google Cloud Data Agent Kit .
2. Разверните раздел КАТАЛОГ .
3. Разверните идентификатор вашего проекта.
4. Развернуть BigQuery .
5. Разверните набор данных transactions_dataset_evals .
6. Щелкните по таблице raw_transactions , чтобы открыть ее подробную информацию в главном редакторе.
7. В левой панели навигации перейдите на вкладки «Данные» , «Схема» и «Подробности» , чтобы просмотреть загруженные записи и метаданные.

Краткое содержание раздела: Вы использовали естественный язык в чате агента для генерации полной рабочей нагрузки Spark Serverless. Затем вы выполнили ее для обработки неструктурированных JSON-логов и преобразования их в таблицу BigQuery (в необработанном виде).

4. Удалите дубликаты и нормализуйте данные с помощью dbt.

Перед обучением модели машинного обучения необходимо обеспечить качество данных, удалив дубликаты потоковых логов, выделив некорректные записи (например, пустые идентификаторы транзакций) и объединив многомерные данные (плательщиков и получателей). Этот процесс требует идемпотентных и надежных SQL-преобразований, поэтому инструмент dbt (инструмент построения данных) отлично подходит для этой цели.

Создайте структуру конвейера DBT.

Используйте агент для создания проекта dbt на основе набора данных BigQuery:

1. Вернитесь в панель чата с агентом .
2. Для создания проекта 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. Агент отобразит документ «План реализации» в главном редакторе. Ознакомьтесь с предлагаемой структурой файлов и логикой SQL-запросов.
4. Нажмите «Продолжить» (а затем «Принять все »), чтобы разрешить агенту сгенерировать файлы в вашем рабочем пространстве.

5. После завершения генерации агент отобразит пошаговое руководство, в котором будут кратко описаны новые компоненты. При появлении запроса подтвердите все изменения.

Интеллектуальное управление зависимостями: при создании шаблонов проектов агент заблаговременно проверяет вашу среду командной строки на наличие необходимых инструментов CLI (таких как dbt ). Если они отсутствуют, агент может автоматически предложить создать виртуальную среду и установить dbt-bigquery в фоновом режиме, запросив ваше подтверждение перед выполнением команд настройки.

Сборка и тестирование

Хотя агент автоматически запустил dbt compile для проверки синтаксической корректности сгенерированного SQL-запроса, теперь вам предстоит материализовать эти представления и таблицы в BigQuery и запустить тесты качества данных для локальной проверки. (Примечание: позже в лабораторной работе вы автоматизируете этот шаг dbt в рамках сквозного DAG Airflow).

1. В левой панели действий щелкните значок Проводника (или нажмите Cmd/Ctrl+Shift+E ).
2. Разверните dbt_project -> models , чтобы просмотреть сгенерированные SQL-модели. Щелкните enriched_transactions.sql , чтобы открыть и просмотреть логику преобразования и обработки мошеннических операций в редакторе.
3. В проводнике файлов щелкните правой кнопкой мыши папку dbt_project и выберите «Открыть во встроенном терминале» . Это автоматически откроет окно терминала, расположенное непосредственно в нужном рабочем каталоге dbt_project .
4. Если у вас еще не установлен dbt , создайте виртуальное окружение вне dbt_project/ (в корневой директории вашей домашней папки или рабочей области) и установите адаптер BigQuery:

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

5. Запустите модели dbt и соответствующие им тесты качества данных:

dbt build

6. Следите за выводом терминала. dbt скомпилирует SQL-запрос, создаст промежуточные и обогащенные таблицы в BigQuery и выполнит проверку данных.

Устранение неполадок при сборке с помощью агента: Если dbt build возникают ошибки (например, несоответствие схем или отсутствие связей), скопируйте и вставьте сообщение об ошибке непосредственно в чат с агентом . Агент проанализирует сбой, скорректирует модели и устранит проблему.

7. После завершения сборки закройте окно терминала, чтобы освободить место на экране для оставшихся шагов.

Краткое содержание раздела: Вы создали проект dbt с помощью агента, провели тесты качества данных и преобразовали исходные записи в промежуточные и обогащенные таблицы BigQuery.

5. Обучение распределенной модели обнаружения мошенничества с использованием случайного леса.

Используя обогащенные данные о транзакциях, материализованные в BigQuery, вы создадите модель машинного обучения для классификации мошеннических событий. Случайный лес — это метод ансамблевого обучения, хорошо подходящий для классификации табличных данных. Запуск RandomForestClassifier на Spark Serverless распределяет обучение модели между рабочими узлами без необходимости управления инфраструктурой.

На этом этапе вы будете использовать агент для генерации конвейера обучения Spark ML.

Сгенерируйте блокнот для обучения машинному обучению.

1. Откройте панель чата с агентом .
2. Для создания последовательности обучения модели введите следующий запрос (не забудьте заменить ${PROJECT_ID} на идентификатор вашего активного проекта):

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. Просмотрите план агента или сгенерированный код и нажмите «Продолжить» / «Принять все» , чтобы сохранить notebooks/02_training.ipynb в свою рабочую область.

Просмотрите и запустите блокнот.

1. Откройте notebooks/02_training.ipynb в редакторе.
2. Ознакомьтесь с этапами конвейера машинного обучения PySpark, посвященными кодированию признаков, сборке векторов и логике классификации с использованием алгоритма случайного леса.
3. В панели инструментов блокнота IDE нажмите кнопку «Запустить все» .
4. Когда откроется выпадающее меню выбора ядра , выберите fraud-pipeline-runtime в Serverless Spark .

Проверка

После завершения выполнения убедитесь, что модель была обучена и экспортирована корректно:

1. Проверьте результаты анализа ячеек в нижней части блокнота, чтобы убедиться в правильности указанного значения площади под ROC-кривой (AUC).
2. Чтобы убедиться в успешном сохранении артефактов модели в GCS, разверните панель обозревателя STORAGE на боковой панели Data Agent Kit.
3. Найдите раздел, заканчивающийся на -models (связанный с вашим активным идентификатором проекта), разверните его и проверьте наличие каталога fraud_model и его этапов конвейера.

От блокнота к продакшену: хотя интерактивные блокноты отлично подходят для экспериментов и разработки моделей, ваш агент также может преобразовывать код блокнота в параметризуемые автономные скрипты PySpark для периодического пакетного выполнения на Dataproc Serverless.

Краткое содержание раздела: Вы использовали агент для создания конвейера обучения PySpark ML, обучили модель случайного леса на обогащенной таблице BigQuery и экспортировали модель в Cloud Storage.

6. Пакетный вывод и запись в Cloud Spanner.

Используя обученную прогностическую модель, хранящуюся в Cloud Storage, вы будете запускать пакетный анализ новых транзакций, поступающих через BigQuery. Транзакции с высоким риском необходимо направлять в операционную систему для их проверки группой по соблюдению нормативных требований. Cloud Spanner предоставляет масштабируемую транзакционную базу данных для этой очереди проверки.

Сгенерируйте блокнот для пакетного вывода результатов.

Используйте агент для создания блокнота для вывода результатов, объединяющего BigQuery, Cloud Storage и Cloud Spanner:

1. Откройте панель чата с агентом .
2. Введите следующую подсказку:

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. Подтвердите создание блокнота, чтобы сохранить файл notebooks/03_inference.ipynb в свою рабочую область.

Почему 50% — это высокий риск: Хотя 50% звучит как подбрасывание монеты, подавляющее большинство легитимных транзакций имеют показатель ниже 5%. Показатель 50% и выше отражает 2% наиболее аномальных транзакций в сети, отфильтровывая «шум», чтобы аудиторы проверяли только действительно рискованные события.

Просмотрите и запустите блокнот.

1. Откройте созданный файл notebooks/03_inference.ipynb в редакторе.
2. Ознакомьтесь с последовательностью выполнения операций вывода PySpark:
   • Зависимости: Шаблон Serverless Runtime предоставляет необходимые JAR-зависимости cloud-spanner для выполнения Spark.
   • Форматирование данных: Перед записью скрипт удаляет сложные векторные столбцы Spark ML (такие как исходные признаки и вероятности), чтобы привести данные в соответствие со схемой таблицы Spanner.
   • Spanner Connector: Он записывает отмеченные строки, используя .format("cloud-spanner") для добавления непосредственно в очередь проверки.
3. В панели инструментов блокнота IDE нажмите кнопку «Запустить все» .
4. При запросе на выбор ядра выберите fraud-pipeline-runtime в Serverless Spark .

Проверка

После завершения обработки блокнота вывода вы можете напрямую обращаться к операционной базе данных Spanner внутри IDE:

1. В панели действий IDE откройте панель Google Cloud Data Agent Kit .
2. Разверните раздел КАТАЛОГ .
3. Разверните идентификатор вашего проекта, затем разверните Spanner .
4. Перейдите в cymbal-fraud -> fraud-db -> Tables -> SparkEvalFraudReviewQueue .
5. Щелкните правой кнопкой мыши по таблице и выберите «Запрос к таблице» , затем выполните запрос:

SELECT *
FROM `SparkEvalFraudReviewQueue`
LIMIT 100;

6. В панели «Результаты запроса» ниже вы увидите недавно добавленные строки, представляющие транзакции высокого риска, помеченные для ручной проверки.

Краткое содержание раздела: Вы использовали агент для создания блокнота для пакетного вывода результатов, оценивали немаркированные записи BigQuery с помощью обученной модели и записывали транзакции высокого риска непосредственно в Cloud Spanner.

7. Организуйте и управляйте процессом с помощью Managed Airflow.

В настоящее время ваш конвейер обработки данных состоит из отдельных этапов: блокнота для загрузки данных, проекта преобразования dbt и блокнота для пакетного вывода данных. Чтобы подготовить его к работе в производственной среде, вы объедините их в запланированный граф зависимостей.

Управляемая служба для Apache Airflow (ранее известная как Cloud Composer) предоставляет управляемый механизм оркестровки для этого рабочего процесса. Комплект агентов данных включает функцию конвейеров оркестровки , которая преобразует декларативные определения конвейеров YAML непосредственно в DAG-графы Airflow.

Определите конвейер обработки данных

Используйте агент для генерации конфигурации конвейера оркестрации:

1. В чате с агентом введите следующую подсказку (не забудьте заменить ${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.

Проверьте конфигурацию DAG.

Инструмент Data Agent Kit Orchestrator использует декларативные конфигурации YAML для определения и развертывания конвейеров в Apache Airflow, что позволяет контролировать версии определений и развертывать их с помощью CI/CD.

В панели «Проводник IDE» просмотрите два файла конвейера, созданных агентом в корневой папке вашей рабочей области:

1. deployment.yaml : Откройте этот файл. Он служит реестром вашей среды. Он сопоставляет ваш логический конвейер dev со средой cymbal-airflow , устанавливает регион выполнения ( us-central1 ) и определяет хранилище artifact_storage , куда помещаются скомпилированные DAG-файлы и зависимости.
2. fraud_analysis_pipeline.yaml : Откройте этот файл. Он определяет граф выполнения. В нем задается расписание запуска ( interval: '0 0 * * *' ) и последовательность трех шагов в блоке actions :
   • Действие notebook для загрузки данных из файла 01_ingestion.ipynb , выполняемое в среде Dataproc Serverless.
   • Действие pipeline преобразования, нацеленное на каталог dbt_project , с зависимостью dependsOn , указывающей на этап приема данных.
   • Действие notebook вывода для файла 03_inference.ipynb с зависимостью dependsOn , указывающей на шаг dbt, включающее свойство JAR Spanner.
3. Агент также сведет воедино информацию об этих сгенерированных артефактах на вкладке «Пошаговое руководство» в панели редактора, где будут описаны выполненные настройки и проверки.

Интерактивная конфигурация DAG

Data Agent Kit отображает конфигурацию вашего конвейера в виде интерактивного визуального графа для проверки и редактирования свойств DAG Airflow.

1. В панели действий IDE откройте панель Google Cloud Data Agent Kit .
2. В разделе DATA ENGINEERING разверните Orchestration Pipelines .
3. Щелкните файл fraud_analysis_pipeline.yaml , чтобы открыть холст визуального DAG в главном редакторе.

4. Щелкните узел Schedule trigger вверху. Справа откроется всплывающее окно конфигурации, отображающее разобранную строку Cron ( 0 0 * * * ) и позволяющее настроить такие параметры, как заполнение и синхронизация.
5. Щелкните любой из узлов задач блокнота (например, этап ввода данных или вывода результатов). Всплывающее меню обновится, отображая конкретные сопоставления выполнения Dataproc Serverless и свойства коннектора.
6. Обратите внимание на гиперссылку с именем файла блокнота (например, 01_ingestion.ipynb ) внутри блока узла. Щелкнув по ней, вы откроете блокнот непосредственно в редакторе.
7. В левой боковой панели под разделом «Конвейеры оркестровки» нажмите Deployment configuration . В этом окне отображаются целевой кластер среды dev и выходные артефакты сегмента GCS.

Краткое содержание раздела: Вы создали конфигурацию конвейера оркестрации с помощью агента, определив зависимости между задачами приема данных, dbt и вывода результатов на интерактивном визуальном холсте.

8. Развертывание, выполнение и мониторинг.

После определения локальной группы доступности ресурсов (DAG) вы подключитесь к управляемой среде Airflow, созданной во время настройки, и развернете конвейер.

Настройка управляемой службы для Apache Airflow

Перед развертыванием настройте подключение к планировщику в параметрах Data Agent Kit таким образом, чтобы расширение было ориентировано на вашу среду Managed Airflow:

1. В панели действий IDE откройте панель Google Cloud Data Agent Kit .
2. В разделе SETTINGS нажмите «Настройки» .
3. Выберите «Планировщик» в левом меню.
4. Настройте параметры:
   • Идентификатор проекта : Выберите идентификатор вашего активного проекта.
   • Регион : Выберите us-central1 .
   • Окружение : Выберите cymbal-airflow .
5. Нажмите « Сохранить ».

Разверните DAG

Оптимизация артефактов развертывания: Перед развертыванием убедитесь, что временные папки, такие как .venv , target/ или logs/ не вложены в каталог dbt_project/ . Загрузка больших виртуальных сред, содержащих тысячи файлов, может привести к зависанию процесса упаковки и замедлению анализа DAG Airflow. Если они присутствуют, удалите их ( rm -rf dbt_project/.venv dbt_project/target dbt_project/logs ) или добавьте в файл .airflowignore .

Теперь вы сможете развернуть настроенный конвейер непосредственно в вашу среду Managed Airflow с помощью визуального холста:

1. В боковой панели Google Cloud Data Agent Kit разверните раздел DATA ENGINEERING > Orchestration Pipelines и щелкните файл fraud_analysis_pipeline.yaml , чтобы открыть холст визуального DAG.
2. В правом верхнем углу панели инструментов холста нажмите синюю кнопку «Запустить конвейер» .
3. В раскрывающемся списке выбора среды выберите dev .
4. Следите за уведомлением о ходе выполнения в нижней области состояния ( Running pipeline: Building pipeline locally... ). Расширение автоматически скомпилирует ваш DAG, упакует ресурсы ноутбука и dbt и загрузит их в хранилище GCS вашей среды Managed Airflow (это займет около 3–4 минут).

Следите за ходом

После завершения локальной компиляции и появления всплывающего уведомления с подтверждением Triggered a new run for pipeline... successfully , отслеживайте выполнение в реальном времени:

1. В боковой панели Google Cloud Data Agent Kit разверните раздел DATA ENGINEERING > Orchestration Pipelines .
2. Нажмите «Управление конвейерами» .
3. В таблице «Управление конвейерами» щелкните по fraud_analysis_pipeline , чтобы открыть историю его выполнения.

4. В окне «История выполнения» выберите активный запуск из календаря.
5. По мере выполнения каждой задачи конвейера (загрузка данных, преобразование dbt и вывод результатов) обновляются индикаторы состояния и отображается продолжительность выполнения задач. Щелкните любую задачу, чтобы просмотреть результаты ее выполнения в режиме реального времени и журналы Airflow DAG.

Развертывание в производственной среде с использованием CI/CD и GitHub Actions: В дополнение к локальному развертыванию, Data Agent Kit автоматически генерирует рабочие процессы GitHub Actions ( .github/workflows/deploy.yaml и .github/workflows/validate.yaml , легко адаптируемые для GitLab CI или Cloud Build). Для рабочих процессов в производственной среде фиксация кода и отправка или слияние изменений в main ветку вашего репозитория GitHub автоматически запускает GitHub Action для проверки и развертывания конвейера в вашей производственной среде Managed Airflow.

Краткое содержание раздела: Вы настроили подключение к планировщику Airflow, развернули сквозной аналитический конвейер в управляемом Airflow и отслеживали выполнение в реальном времени, проверяя систему от необработанных логов до окончательных прогнозов Cloud Spanner.

9. Уборка

Чтобы избежать постоянных расходов на ресурсы, используемые в этом практическом занятии, для вашего проекта Google Cloud, удалите среду с помощью автоматического скрипта.

Не пропускайте этап завершения процесса! За использование Managed Airflow и Cloud Spanner взимается постоянная плата в течение всего периода развертывания. Если вы пропустите этап завершения процесса, выполните его позже или удалите весь проект Google Cloud, созданный для этой лабораторной работы.

1. В панели Терминала (или в Cloud Shell) перейдите в каталог scripts и выполните следующую команду:

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

2. Скрипт выведет список всех ресурсов, которые планирует удалить, и запросит подтверждение:
   • Управляемая среда воздушного потока ( cymbal-airflow )
   • Экземпляр Cloud Spanner ( cymbal-fraud )
   • Набор данных BigQuery ( transactions_dataset_evals )
   • Сегменты облачного хранилища ( gs://${PROJECT_ID}-fin-clearing-raw и gs://${PROJECT_ID}-models )
   • Счет для оплаты услуг работника ( composer-worker-sa )
3. Введите y для подтверждения. Скрипт завершения удалит все настроенные службы GCP и очистит локальные файлы.

10. Поздравляем!

Вы создали комплексный конвейер обнаружения мошенничества, охватывающий Cloud Storage, BigQuery, управляемый сервис для Apache Spark (Spark Serverless), dbt, Cloud Spanner и управляемый сервис для Apache Airflow, используя парное программирование с Google Cloud Data Agent Kit внутри среды разработки Antigravity IDE.

Чего вы достигли

1. 📥 Ввод необработанных журналов транзакций в таблицу BigQuery с использованием Managed Service for Apache Spark и Data Agent Kit.
2. 🧹 Данные были дедуплицированы и нормализованы путем создания проекта dbt с тестами качества данных.
3. 🤖 Обучил распределенную модель случайного леса с помощью RandomForestClassifier и экспортировал обученную модель в облачное хранилище.
4. ⚡ Выполнен пакетный анализ входящих транзакций, и записи с высоким риском перенаправлены в Cloud Spanner для проверки в рамках аудита.
5. 🔄 Организовал, развернул и отслеживал рабочий процесс в виде запланированного DAG-графа Airflow, используя управляемую службу для Apache Airflow и инструменты визуального управления DAG в IDE.

Ключевые понятия

Концепция	Что вы узнали
Комплект агента данных	Парное программирование внутри IDE с использованием естественного языка для генерации блокнотов PySpark, настройки моделей dbt и определения DAG-графов Airflow.
BigQuery	Масштабируемое табличное хранилище для аналитического SQL, преобразований dbt и обучения машинного обучения.
Spark Serverless	Бессерверное выполнение для распределенной загрузки данных PySpark и обучения алгоритма машинного обучения Random Forest
Коннектор Cloud Spanner	Запись пакетных прогнозов Spark непосредственно в очереди проверки операционной базы данных.
Заявления YAML DAG	Декларативные определения конвейеров, отображаемые в виде интерактивных визуальных графов Airflow в IDE.
Визуальное управление DAG	Проверка зависимостей конвейера, развертывание в Managed Airflow и мониторинг истории выполнения задач в режиме реального времени внутри IDE.

Следующие шаги

• Ознакомьтесь с документацией по Google Cloud Data Agent Kit.
• Узнайте больше об управляемых сервисах для Apache Spark.
• Узнайте больше об управляемых сервисах для Apache Airflow.
• Создавайте собственные многосервисные конвейеры с помощью Antigravity IDE.