Building Real-World AI Agent on GCP — Data Pipeline: CDC, ETL & Orchestration

Intro.

In Part 1, we designed the Walsh Retail AI Agent architecture, set up Cloud SQL as the operational database (stage_db), and created the BigQuery schema (ai_agent_db). By the end of Part 1, we had a clean database layer ready to serve as the foundation for the entire pipeline.

In Part 2, we build the Data Engineering Pipeline — the layer that moves data from the operational system into the analytical layer in real time, transforms it into ML-ready features, and orchestrates the entire process automatically. This is where the architecture becomes a live, running system.

0. Pipeline Overview

The diagram below shows the full Part 2 data engineering pipeline. Cloud SQL (completed in Part 1) feeds data into the pipeline via Datastream CDC. Cloud Workflows orchestrates the entire pipeline — from CDC through ETL to the final ML-ready feature store.

Figure 0. Walsh Retail Part 2 — Data Engineering Pipeline Overview

1. What Is a Data Pipeline — and Why Do We Need One?

A data pipeline is an automated system that moves data from one place to another — collecting, transforming, and loading it so it is always available in the right format at the right time. Without a pipeline, data sits in the operational database and never reaches the analytical tools that need it.

In the Walsh Retail system, the pipeline has three jobs:

• Capture — Move data from Cloud SQL to BigQuery in real time (CDC)
• Transform — Convert raw transactional data into ML-ready features (ETL)
• Automate — Run the entire process automatically on a daily schedule (Orchestration)

2. Pipeline Architecture Overview

The Part 2 pipeline consists of four layers working in sequence:

Data flows in one direction: Cloud SQL → Datastream → BigQuery (ecommerce_db) → Dataform → BigQuery (mart) → Vertex AI (Part 3)

3. Datastream CDC — Real-Time Change Data Capture

3.1 What Is CDC and Why Datastream?

Change Data Capture (CDC) is a technique that captures every insert, update, and delete from a source database and streams those changes to a destination in real time — without batch jobs, without manual exports, and without impacting the operational system.

Datastream reads from Cloud SQL's binary log (binlog), which MySQL writes automatically for every committed transaction. Three properties make this the right choice:

• Resilient — if interrupted, it resumes exactly where it left off. No data gap, no duplicate load.
• Responsive — changes appear in BigQuery within seconds of the source transaction, not hours.
• Simultaneous — the operational database continues serving transactions while CDC streams in the background, invisibly.

3.2 Prerequisites — Enabling Binary Log

Datastream requires MySQL binary logging. For Cloud SQL Enterprise edition, enable Point-in-time Recovery (PITR) to activate binary logging, and set the retention flag:

Navigation: GCP Console > SQL > ai-db > Edit > Flags > Add flag: binlog_expire_logs_seconds = 604800 > Save

Figure 3.1. Cloud SQL ai-db — binlog_expire_logs_seconds = 604800 Settings

3.3 Datastream IP Allowlisting

Before creating the connection profile, add Datastream's IP addresses to Cloud SQL's Authorized Networks:

• 34.145.160.237/32
• 35.245.110.238/32
• 34.150.213.121/32
• 34.85.182.28/32
• 34.150.171.115/32

Navigation: GCP Console > SQL > ai-db > Connections > Networking > Authorized networks > Add network > Save

Figure 3.2. Cloud SQL Authorized Networks — 5 Datastream IPs registered

3.4 Connection Profiles

Two connection profiles define how Datastream connects to the source and destination:

Navigation: GCP Console > Datastream > Connection profiles > Create profile > MySQL > Enter settings > Test > Create

Figure 3.3. Datastream — walsh-cloudsql-source Connection test passed

Figure 3.4. Datastream — walsh-cloudsql-source profile created

Figure 3.5. Datastream — walsh-bigquery-dest profile created

3.5 Creating and Starting the Stream

Navigation: GCP Console > Datastream > Streams > Create stream > Configure > Select source/dest profiles > Select tables > Create & Start

Figure 3.6. walsh-retail-stream — Stream started successfully

Figure 3.7. walsh-retail-stream — RUNNING, Throughput ~176 events/sec on initial backfill

Figure 3.8. BigQuery ecommerce_db — CDC data loaded with datastream_metadata column

4. Dataform ETL — OLTP to OLAP Transformation

4.1 Why Dataform?

Raw CDC data in BigQuery mirrors the operational schema — normalized, transaction-optimized, and not suitable for ML training directly. Dataform transforms this raw data into a denormalized, analytics-ready feature table using SQLX — SQL with metadata config blocks that define transformations as versioned, testable assets native to BigQuery.

4.2 What Is SQLX?

SQLX is Dataform's file format — it's standard SQL with a config {} block at the top that tells Dataform what to create and where to put it. Think of it as SQL + deployment instructions in one file.

A SQLX file has two parts:

When you click Start execution in the Dataform workspace, it compiles this SQLX into a BigQuery job and runs CREATE OR REPLACE TABLE isupernova-ai.mart.customer_features AS (SELECT ...) automatically. No manual DDL needed.

4.2 RFM Feature Engineering

The core transformation produces mart.customer_features — one row per customer with 10 RFM features for the Random Forest model:

4.3 Dataform Setup & Execution

Setting up Dataform takes three steps: create a repository, create a workspace inside it, and write the SQLX file. The workspace is where you edit and test SQLX files before running them.

• Repository: walsh-retail-dataform (us-east4) — the top-level container
• Workspace: walsh-retail-workspace — your editable development environment
• File: definitions/customer_features.sqlx — the transformation definition

Once the SQLX file is saved, click Start execution to trigger the compilation and run the BigQuery job. Dataform compiles the SQLX into a CREATE OR REPLACE TABLE statement and executes it in BigQuery automatically.

Navigation: GCP Console > Dataform > Create repository: walsh-retail-dataform > Create workspace: walsh-retail-workspace > definitions/ > + > customer_features.sqlx > Start execution

4.4 RFM Feature Engineering Results

Figure 4.1. Dataform workspace — customer_features.sqlx SQLX code + Query Results preview

Figure 4.2. Dataform execution — Status: Success in 4 seconds

Figure 4.3. BigQuery mart.customer_features — 200 customers, all RFM features populated

4.5 ml_prediction Table

A second mart table is created as an empty schema, ready to receive Vertex AI predictions in Part 3:

Figure 4.4. BigQuery mart.ml_prediction — Schema ready for Part 3

5. Cloud Workflows — Pipeline Orchestration

5.1 Why Cloud Workflows?

5.2 How Cloud Workflows Orchestrates Dataform

Cloud Workflows uses a YAML-based definition to connect GCP services via HTTP calls. Each step in the workflow maps to one API call. The default sample code that comes with Cloud Workflows is completely replaced with our custom pipeline definition.

The walsh-retail-pipeline workflow has four steps:

The workflow runs in two phases:

• Compile — createCompilationResult reads all SQLX files in the workspace and compiles them into a versioned snapshot. This is like a "build" step that validates the SQL before running it.
• Execute — createWorkflowInvocation takes the compiled result and submits it to BigQuery as a job. Every SQLX file in the definitions/ folder runs in dependency order.

5.3 IAM Configuration

A dedicated service account dataform-runner@isupernova-ai.iam.gserviceaccount.com is created with minimum required permissions:

• roles/dataform.editor — create and execute Dataform workflow invocations
• roles/bigquery.jobUser — submit BigQuery jobs
• roles/bigquery.dataEditor — read and write BigQuery tables

5.4 Workflow Execution

Navigation: GCP Console > Workflows > Create: walsh-retail-pipeline > Deploy > Execute

Figure 5.1. Cloud Workflows — walsh-retail-pipeline execution Succeeded in 1.004 seconds

6. Cloud Scheduler — Daily Automation

Navigation: GCP Console > Cloud Scheduler > Create job > Name: walsh-retail-daily > Frequency: 0 1 * * * > Target: Workflows > walsh-retail-pipeline > Create

Figure 6.1. Cloud Scheduler — walsh-retail-daily Force run: Success

7. End-to-End Pipeline Verification

The full pipeline was verified end-to-end with a live UPDATE test — tracing a single data change from Cloud SQL all the way through to the final mart feature table:

Figure 7.1. Cloud SQL Studio — CDC test UPDATE executed successfully

Figure 7.2. BigQuery ecommerce_db — dml_flag='U' reflected via Datastream CDC

Figure 7.3. BigQuery Jobs Log — Dataform execution timestamps (INFORMATION_SCHEMA.JOBS)

Figure 7.4. E2E Verification — mart.customer_features JOIN ecommerce_db.customer, update timestamp matched

8. Final Pipeline Diagram

The complete Part 2 Data Engineering Pipeline — from Cloud SQL through Datastream CDC, Dataform ETL, and Cloud Workflows orchestration to the final ML-ready feature store:

Figure 8.1. Walsh Retail Part 2 — Complete Data Engineering Pipeline

With the Data Engineering Pipeline complete — real-time CDC streaming from Cloud SQL to BigQuery, RFM feature transformation via Dataform, and daily automated orchestration via Cloud Workflows and Cloud Scheduler — the Walsh Retail AI Agent now has a continuously updated, ML-ready feature store.

In the next post, we will build the Intelligence Layer — training the Random Forest model on Vertex AI, implementing RAG with BigQuery Vector Search and LangChain, and generating personalized campaign offers with Gemini API. Stay tuned for Part 3.