refactor(terraform): modernize customer data platform guide infrastructure

.agents/skills/use-case-deployment/SKILL.md

@@ -19,7 +19,7 @@ This skill provides step-by-step execution workflows for deploying, running, ver
 | :--- | :--- | :--- | :--- | :--- | :--- |
 | **GenAI & ML** | `terraform/ml_ai` | `pipelines/ml_ai_python` | `./scripts/02_run_dataflow.sh` | Pub/Sub `messages` topic | Pub/Sub `predictions-sub` subscription |
 | **ETL & Integration** | `terraform/etl_integration` | `pipelines/etl_integration_java` | `./scripts/02_run_publisher_dataflow.sh` & `./scripts/03_run_changestream_template.sh` | Pub/Sub Taxirides feed | Cloud Spanner `events` table & BigQuery `replica.events_changelog` |
-| **Customer Data Platform (CDP)** | `terraform/cdp` | `pipelines/cdp` | `./scripts/02_run_dataflow_job.sh` | `python cdp_pipeline/generate_transaction_data.py` | BigQuery `output_dataset.unified-table` |
+| **Customer Data Platform (CDP)** | `terraform/cdp` | `pipelines/cdp` | `./scripts/02_run_dataflow.sh` | `python cdp_pipeline/generate_transaction_data.py` | BigQuery `cdp_dataset.unified_customer_data` |
 | **Anomaly Detection** | `terraform/anomaly_detection` | `pipelines/anomaly_detection` | `./scripts/02_run_dataflow.sh` | Pub/Sub `anomaly-detection-transactions` topic | Pub/Sub `anomaly-detection-detections`, BigQuery `anomaly_detection.detections`, errors `anomaly-detection-errors` |
 | **Marketing Intelligence** | `terraform/marketing_intelligence` | `pipelines/marketing_intelligence` | `./scripts/02_run_dataflow.sh` | Pub/Sub user activity stream | BigQuery marketing attribution tables |
 | **Clickstream Analytics** | `terraform/clickstream_analytics` | `pipelines/clickstream_analytics_java` | `./scripts/01_launch_pipeline.sh` | Pub/Sub events | Cloud Bigtable & BigQuery analytics table |
@@ -98,15 +98,15 @@ terraform init && terraform apply -auto-approve
 
 # 2. Build Container & Launch Dataflow
 cd ../../pipelines/cdp
-source scripts/00_set_variables.sh
-./scripts/01_cloudbuild_and_push_container.sh
-./scripts/02_run_dataflow_job.sh
+source scripts/00_set_environment.sh
+./scripts/01_build_and_push_container.sh
+./scripts/02_run_dataflow.sh
 
 # 3. Generate Streaming Transactions
 python3 ./cdp_pipeline/generate_transaction_data.py
 
 # 4. Validate Unified BigQuery Table
-bq query --use_legacy_sql=false 'SELECT * FROM output_dataset.`unified-table` LIMIT 10'
+bq query --use_legacy_sql=false 'SELECT * FROM cdp_dataset.unified_customer_data LIMIT 10'
 ```
 
 ### 4. Clickstream Analytics with Bigtable (Java)

AGENTS.md

@@ -25,7 +25,7 @@ dataflow-solution-guides/
 ├── terraform/                # Infrastructure-as-Code using Google Cloud Foundation Fabric
 │   ├── ml_ai/                # Pub/Sub topics, Artifact Registry, GCS bucket, Service Account
 │   ├── etl_integration/      # Spanner instance/database/change stream, BigQuery, Service Account
-│   ├── cdp/                  # Pub/Sub topics, BigQuery dataset/tables, VPC
+│   ├── cdp/                  # Pub/Sub topics, BigQuery dataset/table, Artifact Registry, Service Account
 │   ├── anomaly_detection/    # Pub/Sub, Bigtable, BigQuery, Artifact Registry, optional GCS, Worker/training identities (Python-managed Vertex AI workflow)
 │   ├── marketing_intelligence/ # Pub/Sub topics, Firestore, BigQuery dataset, Artifact Registry, Service Account
 │   ├── clickstream_analytics/  # Bigtable instance, Pub/Sub, BigQuery, Service Account

pipelines/cdp/README.md

@@ -1,95 +1,85 @@
 # Customer Data Platform sample pipeline (Python)
 
-This sample pipeline demonstrates how to use Dataflow to process the streaming data in order to build Customer Data platform. We will be reading data form multiple streaming sources, two pub-sub topics in this sample pipeline, will join the data and put it in bigquery table for analytics later on.
+This sample pipeline demonstrates how to use Dataflow to process streaming data in order to build a Customer Data Platform (CDP). It reads data from multiple streaming sources (two Pub/Sub topics: `cdp-transactions` and `cdp-coupon-redemption`), joins the records based on transaction and customer keys, and writes the unified records into a BigQuery table for downstream analytics.
 
-This pipeline is part of the [Dataflow Customer Data Platfrom solution guide](../../use_cases/cdp.md).
+This pipeline is part of the [Dataflow Customer Data Platform solution guide](../../use_cases/CDP.md).
 
 ## Architecture
 
-The generic architecture for an inference pipeline looks like as follows:
+The generic architecture for the CDP pipeline looks as follows:
 
 ![Architecture](../imgs/cdp.png)
 
-In this directory, you will find a specific implementation of the above architecture, with the 
-following stages:
+In this directory, you will find a specific implementation of the above architecture with the following stages:
 
-1. **Data ingestion:** Reads data from a Pub/Sub topic.
-2. **Data preprocessing:** The sample pipeline joins the data from two pub-sub topic based on some key fields. This is to showcase the unification of customer data from different sources to store itin one place.
-3. **Output Data:** The final processed data is then appended to the bigquery table.
+1. **Data ingestion:** Reads streaming records from two Pub/Sub topics (`cdp-transactions` and `cdp-coupon-redemption`).
+2. **Data preprocessing & Unification:** Windows incoming records into fixed 60-second windows and executes a `CoGroupByKey` left join to merge transactions with coupon redemptions based on `(transaction_id, household_key)`.
+3. **Output Data:** Writes unified records into the BigQuery table `cdp_dataset.unified_customer_data`.
 
 ## Selecting the cloud region
 
-Not all the resources may be available in all the regions. The default values included in this
-directory have been tested using `us-central1` as region.
+Not all resources may be available in all regions. The default values included in this directory have been tested using `us-central1` as region.
 
-Moreover, the file `scripts/00_set_variables.sh` specifies a machine type for the Datalow workers.
-The selected machine type, `e2-standard-8`, is the one that we used for unification of data. If that
-type is not available in your region, you can check what machines are available to use with the
-following command:
+Moreover, the environment configuration specifies `e2-standard-8` machine types for the Dataflow workers. If that type is not available in your region, check available machine types using:
 
 ```sh
 gcloud compute machine-types list --zones=<ZONE A>,<ZONE B>,...
 ```
 
-See more info about selecting the right type of machine in the following link:
+See more info about selecting the right type of machine in Google Cloud Compute Engine documentation:
 * https://cloud.google.com/compute/docs/machine-resource
 
 ## How to launch the pipeline
 
-All the scripts are located in the `scripts` directory and prepared to be launched from the top 
-sources directory.
+All scripts are located in the `scripts` directory and prepared to be launched from the `pipelines/cdp` directory.
 
-In the script `scripts/00_set_variables.sh`, define the value of the project id and the region variable:
-
-```
-export PROJECT=<YOUR PROJECT ID>
-export REGION=<YOUR CLOUD REGION>
-```
-
-Leave the rest of variables untouched, although you can override them if you prefer.
-
-After you edit the script, load those variables into the environment
+### 1. Load environment variables
+The environment configuration file `scripts/00_set_environment.sh` is generated automatically when deploying the Terraform infrastructure in `terraform/cdp/`. Load those variables into your current shell:
 
 ```sh
-source scripts/00_set_variables.sh
+source scripts/00_set_environment.sh
 ```
 
-And then run the script that builds and publishes the custom Dataflow container. This container will
-contain all the required dependencies.
+### 2. Build and publish custom container
+Build and push the custom Dataflow worker container to Artifact Registry using Cloud Build:
 
 ```sh
-./scripts/01_cloudbuild_and_push_container.sh
+./scripts/01_build_and_push_container.sh
 ```
 
-This will create a Cloud Build job that can take a few minutes to complete. Once it completes, you
-can trigger the pipeline with the following:
+### 3. Launch Dataflow streaming pipeline
+Submit the streaming pipeline job to Google Cloud Dataflow:
 
 ```sh
-./scripts/02_run_dataflow_job.sh
-```
-You can also directly run below script instead of above 3 steps.
-
-```sh
-./scripts/run.sh
+./scripts/02_run_dataflow.sh
 ```
 
 ## Automated Tests
 
-Execute unit and pipeline tests with `pytest`:
+Execute unit and pipeline transform tests with `pytest`:
 
 ```bash
 pytest tests/ -v
 ```
 
-## Input data
+## Input data simulation
 
-To send data into the pipeline, you need to publish messages in the `transactions` and `coupon-redemption` topics.
-Run the python code below to publish data to these pub-sub topics. This script is reading sample data from GCS buckets and publishing it to the pub-sub topic to create real-time streaming environment for this use case. One can update the GCS bucket location as per their environment. For reference, input files are added to folder ./input_data/.
+To send test data into the pipeline, publish messages to the `cdp-transactions` and `cdp-coupon-redemption` Pub/Sub topics:
 
 ```python3
-./cdp_pipeline/generate_transaction_data.py
+python3 ./cdp_pipeline/generate_transaction_data.py
 ```
 
+This script reads sample transaction and coupon data (either from the configured GCS bucket or from local files in `./input_data/`) and publishes simulated events to the input Pub/Sub topics.
+
 ## Output data
 
-The unified data from the two pub-sub topics is moved to the bigquery table `output_dataset.unified-table`.
+The unified data from the two Pub/Sub topics is stored in the BigQuery table:
+```
+${PROJECT}.${BQ_DATASET}.${BQ_UNIFIED_TABLE}  # Default: cdp_dataset.unified_customer_data
+```
+
+Verify output records via `bq`:
+```bash
+bq query --use_legacy_sql=false "SELECT * FROM \`${PROJECT}.cdp_dataset.unified_customer_data\` LIMIT 10"
+```

pipelines/cdp/cdp_pipeline/generate_transaction_data.py

@@ -15,40 +15,77 @@
 A data generator for the Customer Data Platform analytics pipeline.
 """
 
-from google.cloud import pubsub_v1
+import argparse
+import asyncio
 import json
+import os
+from google.cloud import pubsub_v1
 import pandas as pd
-import asyncio
 
 
-async def publish_coupons_to_pubsub():
-  bucket_name = "<bucket_name>"
-  project_id = "<project_id>"
+async def publish_coupons_to_pubsub(project_id: str | None = None,
+                                    transactions_topic: str | None = None,
+                                    coupons_topic: str | None = None,
+                                    bucket_name: str | None = None):
+  project_id = project_id or os.environ.get("PROJECT", "<project_id>")
+  transactions_topic_name = transactions_topic or os.environ.get(
+      "TRANSACTIONS_TOPIC", "cdp-transactions")
+  coupons_topic_name = coupons_topic or os.environ.get(
+      "COUPON_REDEMPTION_TOPIC", "cdp-coupon-redemption")
+  gcs_bucket_env = os.environ.get("GCS_BUCKET", "")
+  if not bucket_name and gcs_bucket_env:
+    bucket_name = gcs_bucket_env.replace("gs://", "").split("/")[0]
 
-  # Example: ["27601281299","27757099033","28235291311","27021203242","27101290145","27853175697"]
-  transactions_id = [
-      "<List of sample transaction IDs to test the pipeline on.>"
+  sample_transactions_id = [
+      "27601281299", "27757099033", "28235291311", "27021203242",
+      "27101290145", "27853175697"
   ]
-  transactions_topic_name = "transactions"
-  # Reference example - "dataflow-solution-guide-cdp/input_data/transaction_data.csv"
-  transactions_data = "<path to transactions data in gcs bucket>"
 
-  coupons_topic_name = "coupon_redemption"
-  # reference example - "dataflow-solution-guide-cdp/input_data/coupon_redempt.csv"
-  coupons_data = "<path to coupon redemption data in gcs bucket>"
+  # Local directory fallback
+  current_dir = os.path.dirname(os.path.abspath(__file__))
+  local_trans_path = os.path.join(
+      os.path.dirname(current_dir), "input_data", "transaction_data.csv")
+  local_coupons_path = os.path.join(
+      os.path.dirname(current_dir), "input_data", "coupon_redempt.csv")
+
+  if bucket_name:
+    gcs_prefix = (
+        f"gs://{bucket_name}/assets/dataflow-solution-guide-cdp/input_data"
+    )
+    trans_gcs = f"{gcs_prefix}/transaction_data.csv"
+    coupons_gcs = f"{gcs_prefix}/coupon_redempt.csv"
+    try:
+      transactions_df = pd.read_csv(trans_gcs, dtype=str)
+      coupons_df = pd.read_csv(coupons_gcs, dtype=str)
+    except Exception:  # pylint: disable=broad-exception-caught
+      print(
+          f"Falling back to local CSV files from {local_trans_path} and "
+          f"{local_coupons_path}")
+      transactions_df = pd.read_csv(local_trans_path, dtype=str)
+      coupons_df = pd.read_csv(local_coupons_path, dtype=str)
+  else:
+    transactions_df = pd.read_csv(local_trans_path, dtype=str)
+    coupons_df = pd.read_csv(local_coupons_path, dtype=str)
 
-  transactions_df = pd.read_csv(
-      f"gs://{bucket_name}/{transactions_data}", dtype=str)
-  coupons_df = pd.read_csv(f"gs://{bucket_name}/{coupons_data}", dtype=str)
   publisher = pubsub_v1.PublisherClient()
+  transactions_topic_path = (
+      transactions_topic_name
+      if transactions_topic_name.startswith("projects/") else
+      publisher.topic_path(project_id, transactions_topic_name))
+  coupons_topic_path = (
+      coupons_topic_name if coupons_topic_name.startswith("projects/") else
+      publisher.topic_path(project_id, coupons_topic_name))
 
-  transactions_topic_path = publisher.topic_path(project_id,
-                                                 transactions_topic_name)
-  coupons_topic_path = publisher.topic_path(project_id, coupons_topic_name)
   filtered_trans_df = transactions_df[transactions_df["transaction_id"].isin(
-      transactions_id)]
+      sample_transactions_id)]
   filtered_coupons_df = coupons_df[coupons_df["transaction_id"].isin(
-      transactions_id)]
+      sample_transactions_id)]
+
+  if filtered_trans_df.empty:
+    filtered_trans_df = transactions_df
+  if filtered_coupons_df.empty:
+    filtered_coupons_df = coupons_df
+
   await asyncio.gather(
       publish_coupons(filtered_coupons_df, publisher, coupons_topic_path),
       publish_transactions(filtered_trans_df, publisher,
@@ -75,4 +112,31 @@ async def publish_transactions(filtered_trans_df, publisher,
 
 
 if __name__ == "__main__":
-  asyncio.run(publish_coupons_to_pubsub())
+  parser = argparse.ArgumentParser(
+      description="Publish sample transactions and coupons to Pub/Sub.")
+  parser.add_argument(
+      "--project_id",
+      default=os.environ.get("PROJECT"),
+      help="GCP Project ID (defaults to $PROJECT)")
+  parser.add_argument(
+      "--transactions_topic",
+      default=os.environ.get("TRANSACTIONS_TOPIC"),
+      help="Transactions Pub/Sub topic name or ID (defaults to $TRANSACTIONS_TOPIC)"
+  )
+  parser.add_argument(
+      "--coupons_topic",
+      default=os.environ.get("COUPON_REDEMPTION_TOPIC"),
+      help="Coupons Pub/Sub topic name or ID (defaults to $COUPON_REDEMPTION_TOPIC)"
+  )
+  parser.add_argument(
+      "--bucket_name",
+      default=None,
+      help="Optional GCS bucket containing input data")
+  args = parser.parse_args()
+
+  asyncio.run(
+      publish_coupons_to_pubsub(
+          project_id=args.project_id,
+          transactions_topic=args.transactions_topic,
+          coupons_topic=args.coupons_topic,
+          bucket_name=args.bucket_name))

pipelines/cdp/scripts/.gitignore

@@ -0,0 +1 @@
+00_set_environment.sh

pipelines/cdp/scripts/01_build_and_push_container.sh

@@ -0,0 +1,36 @@
+#!/usr/bin/env bash
+#  Copyright 2026 Google LLC
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      https://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+set -e
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PIPELINE_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
+
+if [ -f "$SCRIPT_DIR/00_set_environment.sh" ]; then
+  # shellcheck source=/dev/null
+  source "$SCRIPT_DIR/00_set_environment.sh"
+fi
+
+: "${PROJECT:?PROJECT must be set or source 00_set_environment.sh}"
+: "${REGION:?REGION must be set or source 00_set_environment.sh}"
+: "${CONTAINER_URI:?CONTAINER_URI must be set or source 00_set_environment.sh}"
+
+echo "Building and pushing container image: $CONTAINER_URI..."
+gcloud builds submit \
+  --project="$PROJECT" \
+  --region="$REGION" \
+  --default-buckets-behavior=regional-user-owned-bucket \
+  --substitutions _TAG="$CONTAINER_URI" \
+  "$PIPELINE_DIR"