diff --git a/.agents/skills/use-case-deployment/SKILL.md b/.agents/skills/use-case-deployment/SKILL.md index 6618b954..8a2eb70f 100644 --- a/.agents/skills/use-case-deployment/SKILL.md +++ b/.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) diff --git a/AGENTS.md b/AGENTS.md index f705ff7e..82b95f80 100644 --- a/AGENTS.md +++ b/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 diff --git a/pipelines/cdp/README.md b/pipelines/cdp/README.md index dd9be0f5..8355668b 100644 --- a/pipelines/cdp/README.md +++ b/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=,,... ``` -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= -export 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`. \ No newline at end of file +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" +``` \ No newline at end of file diff --git a/pipelines/cdp/cdp_pipeline/generate_transaction_data.py b/pipelines/cdp/cdp_pipeline/generate_transaction_data.py index 2ec9893c..7e800d1c 100644 --- a/pipelines/cdp/cdp_pipeline/generate_transaction_data.py +++ b/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 = "" - 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", "") + 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 = [ - "" + 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 = "" - coupons_topic_name = "coupon_redemption" - # reference example - "dataflow-solution-guide-cdp/input_data/coupon_redempt.csv" - coupons_data = "" + # 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)) diff --git a/pipelines/cdp/scripts/.gitignore b/pipelines/cdp/scripts/.gitignore new file mode 100644 index 00000000..bb35710f --- /dev/null +++ b/pipelines/cdp/scripts/.gitignore @@ -0,0 +1 @@ +00_set_environment.sh diff --git a/pipelines/cdp/scripts/01_build_and_push_container.sh b/pipelines/cdp/scripts/01_build_and_push_container.sh new file mode 100755 index 00000000..6bc5e8f0 --- /dev/null +++ b/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" diff --git a/pipelines/cdp/scripts/01_cloudbuild_and_push_container.sh b/pipelines/cdp/scripts/01_cloudbuild_and_push_container.sh deleted file mode 100644 index 845561bc..00000000 --- a/pipelines/cdp/scripts/01_cloudbuild_and_push_container.sh +++ /dev/null @@ -1,5 +0,0 @@ -gcloud builds submit \ - --region=$REGION \ - --default-buckets-behavior=regional-user-owned-bucket \ - --substitutions _TAG=$CONTAINER_URI \ - . diff --git a/pipelines/cdp/scripts/02_run_dataflow.sh b/pipelines/cdp/scripts/02_run_dataflow.sh new file mode 100755 index 00000000..165a47b4 --- /dev/null +++ b/pipelines/cdp/scripts/02_run_dataflow.sh @@ -0,0 +1,57 @@ +#!/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)" + +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}" +: "${SERVICE_ACCOUNT:?SERVICE_ACCOUNT must be set or source 00_set_environment.sh}" +: "${CONTAINER_URI:?CONTAINER_URI must be set or source 00_set_environment.sh}" + +SUBNET_OPT="" +if [ -n "$SUBNETWORK" ]; then + SUBNET_OPT="--subnetwork=$SUBNETWORK" +elif [ -n "$NETWORK" ]; then + SUBNET_OPT="--subnetwork=$NETWORK" +fi + +echo "Submitting Customer Data Platform Dataflow pipeline..." +python3 -m main \ + --streaming \ + --runner=DataflowRunner \ + --project="$PROJECT" \ + --temp_location="${TEMP_LOCATION:-gs://$PROJECT/tmp}" \ + --region="$REGION" \ + --save_main_session \ + --service_account_email="$SERVICE_ACCOUNT" \ + $SUBNET_OPT \ + --no_use_public_ips \ + --sdk_container_image="$CONTAINER_URI" \ + --max_num_workers="$MAX_DATAFLOW_WORKERS" \ + --disk_size_gb="$DISK_SIZE_GB" \ + --machine_type="$MACHINE_TYPE" \ + --transactions_topic="$TRANSACTIONS_TOPIC" \ + --coupons_redemption_topic="$COUPON_REDEMPTION_TOPIC" \ + --output_dataset="$BQ_DATASET" \ + --output_table="$BQ_UNIFIED_TABLE" \ + --project_id="$PROJECT" \ + --enable_streaming_engine diff --git a/pipelines/cdp/scripts/02_run_dataflow_job.sh b/pipelines/cdp/scripts/02_run_dataflow_job.sh deleted file mode 100644 index 2168f0dc..00000000 --- a/pipelines/cdp/scripts/02_run_dataflow_job.sh +++ /dev/null @@ -1,19 +0,0 @@ -python3 -m main \ - --streaming \ - --runner=DataflowRunner \ - --project=$PROJECT \ - --temp_location=gs://$PROJECT/tmp \ - --region=$REGION \ - --save_main_session \ - --service_account_email=$SERVICE_ACCOUNT \ - --subnetwork=$SUBNETWORK \ - --sdk_container_image=$CONTAINER_URI \ - --max_workers=$MAX_DATAFLOW_WORKERS \ - --disk_size_gb=$DISK_SIZE_GB \ - --machine_type=$MACHINE_TYPE \ - --transactions_topic=$TRANSACTIONS_TOPIC \ - --coupons_redemption_topic=$COUPON_REDEMPTION_TOPIC \ - --output_dataset=$BQ_DATASET \ - --output_table=$BQ_UNIFIED_TABLE \ - --project_id=$PROJECT \ - --enable_streaming_engine diff --git a/pipelines/cdp/scripts/run.sh b/pipelines/cdp/scripts/run.sh deleted file mode 100644 index 77ced02f..00000000 --- a/pipelines/cdp/scripts/run.sh +++ /dev/null @@ -1,3 +0,0 @@ -source ./scripts/00_set_variables.sh -sh ./scripts/01_cloudbuild_and_push_container.sh -sh ./scripts/02_run_dataflow_job.sh \ No newline at end of file diff --git a/terraform/AGENTS.md b/terraform/AGENTS.md index 92e9096c..2e786566 100644 --- a/terraform/AGENTS.md +++ b/terraform/AGENTS.md @@ -10,7 +10,7 @@ This directory contains Terraform infrastructure definitions for each solution g | :--- | :--- | :--- | | `ml_ai/` | Pub/Sub Topics (`messages`, `predictions`), Artifact Registry (`dataflow-containers`), GCS Bucket, Service Account | `pipelines/ml_ai_python/` | | `etl_integration/` | Cloud Spanner (taxis DB + Change Stream), BigQuery Dataset, Service Account | `pipelines/etl_integration_java/` | -| `cdp/` | Pub/Sub Topics (`transactions`, `coupon-redemption`), BigQuery Dataset, VPC, Subnet | `pipelines/cdp/` | +| `cdp/` | Pub/Sub Topics (`cdp-transactions`, `cdp-coupon-redemption`), BigQuery Dataset (`cdp_dataset`) & Table (`unified_customer_data`), Artifact Registry (`cdp-containers`), Service Account (`cdp-dataflow-sa`) | `pipelines/cdp/` | | `anomaly_detection/` | Pub/Sub, Bigtable, BigQuery, Artifact Registry, optional GCS, Worker and training identities; Python-managed Vertex AI workflow | `pipelines/anomaly_detection/` | | `marketing_intelligence/` | Pub/Sub Topics (`input`, `output`), Cloud Firestore (Native Mode), BigQuery Dataset, Artifact Registry, Service Account | `pipelines/marketing_intelligence/` | | `clickstream_analytics/` | Cloud Bigtable (Instance & Table), Pub/Sub Topic, BigQuery Dataset, Service Account | `pipelines/clickstream_analytics_java/` | diff --git a/terraform/TERRAFORM_REVAMP_GUIDE.md b/terraform/TERRAFORM_REVAMP_GUIDE.md index bf7880e5..bdff5139 100644 --- a/terraform/TERRAFORM_REVAMP_GUIDE.md +++ b/terraform/TERRAFORM_REVAMP_GUIDE.md @@ -321,7 +321,7 @@ pylint --rcfile ../pylintrc . | # | Use Case / Directory | Key Application Resources to Keep | Recommended Default SA Name | Target Pipeline | | :- | :--- | :--- | :--- | :--- | | 1 | **`etl_integration/`** *(Done)* | Spanner Instance/DBs/IAM, BigQuery Dataset | `spanner-cdc-dataflow-sa` | `pipelines/etl_integration_java/` | -| 2 | **`cdp/`** | Pub/Sub Topics (`transactions`, `coupon_redemption`), BigQuery Dataset, Artifact Registry | `cdp-dataflow-sa` | `pipelines/cdp/` | +| 2 | **`cdp/`** *(Done)* | Pub/Sub Topics (`cdp-transactions`, `cdp-coupon-redemption`), BigQuery Dataset (`cdp_dataset`) & Table (`unified_customer_data`), Artifact Registry (`cdp-containers`) | `cdp-dataflow-sa` | `pipelines/cdp/` | | 3 | **`clickstream_analytics/`** *(Done)* | Cloud Bigtable (Instance & Table), Pub/Sub Topic, BigQuery Dataset & Tables | `clickstream-dataflow-sa` | `pipelines/clickstream_analytics_java/` | | 4 | **`anomaly_detection/`** | Pub/Sub Topic, BigQuery Dataset, Vertex AI Endpoint | `anomaly-detection-sa` | `pipelines/anomaly_detection/` | | 5 | **`marketing_intelligence/`** *(Done)* | Pub/Sub Topics, BigQuery Dataset, Firestore Native, Artifact Registry | `marketing-intel-sa` | `pipelines/marketing_intelligence/` | diff --git a/terraform/cdp/README.md b/terraform/cdp/README.md index 4c6bf78f..c72ed1c6 100644 --- a/terraform/cdp/README.md +++ b/terraform/cdp/README.md @@ -1,93 +1,97 @@ -# Customer Data Platform project deployment +# Customer Data Platform (CDP) Infrastructure Deployment -This directory contains the Terraform code to spawn a Google Cloud project -with all the necessary infrastructure and configuration required for running -the Customer Data Platform solution guide. +This directory contains the Terraform code to provision application-level infrastructure and configuration required for running the Customer Data Platform solution guide on Google Cloud. -These deployment scripts are part of the -[Dataflow Customer Data Platform](../../use_cases/cdp.md). +These deployment scripts are part of the [Dataflow Customer Data Platform Solution Guide](../../use_cases/CDP.md). ## Bill of resources created by this script -The scripts will create the following resources - -| Resource | Name | Description | -| :--------------- | :-------------------: | :----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Project | Set by user | Optional, you may reuse an existing project. If the project is created by Terraform, it will enable the APIs for Cloud Build, Dataflow, Monitoring, Pub/Sub, Dataflow Autoscaling, and Artifact Registry | -| Docker registry | `dataflow-containers` | An Artifact Registry Docker repo for the custom Dataflow container used in the pipeline. The Cloud Build service agent is granted admin role in this repository. The Dataflow service account is granted reader role. By default, only the 3 latest versions of each image are kept in the repo. | -| Bucket | Same as project id | Using the standard storage class, this is a regional bucket in the region specified by the user. | -| Pub/Sub topic | `transactions` | The first input Pub/Sub topic for the sample pipeline. | -| Pub/Sub topic | `coupon_redemption` | The second input Pub/Sub topic for the sample pipeline. | -| Bigquery Dataset | `output_dataset` | The output dataset for the sample pipeline. | -| Service account | `my-dataflow-sa` | Dataflow worker service account. It has storage admin, Dataflow worker, metrics writer and Pub/Sub editor roles assigned at project level. | -| VPC network | `dataflow-net` | If the project is created from scratch, the default network is removed and this network is re-created with a single regional sub-network. | -| Cloud NAT | `dataflow-nat` | Cloud NAT in the region specified by the user, in case the Dataflow workers need to reach the Internet. This is not necessary for the sample pipeline provided. | -| Firewall rules | Several rules | Ingress and egress rules to remove unnecessary traffic, and to ensure the traffice required by Dataflow. If you want to access a VM using SSH, apply the network tag `ssh` to that instance. Same for `http-server` and for `https-server` | +The scripts will create the following application-level resources: -## Configuration variables - -This deployment accepts the following configuration variables: +| Resource | Name | Description | +| :--- | :---: | :--- | +| **Docker registry** | `cdp-containers` | An Artifact Registry Docker repository for the custom Dataflow container image. Cloud Build is granted admin role and the Dataflow worker service account is granted reader role. By default, the 3 latest image versions are retained. | +| **GCS Bucket** *(Optional)* | `var.bucket_name` or `var.project_id` | Optional standard regional GCS bucket for Dataflow temp and staging files (created when `create_bucket = true`). | +| **Pub/Sub topic** | `cdp-transactions` | The first input Pub/Sub topic for streaming customer transaction events. | +| **Pub/Sub topic** | `cdp-coupon-redemption` | The second input Pub/Sub topic for streaming coupon redemption events. | +| **BigQuery Dataset** | `cdp_dataset` | The destination BigQuery dataset for customer data unification. | +| **BigQuery Table** | `unified_customer_data` | The destination BigQuery table storing joined transaction and coupon redemption records. | +| **Service Account** | `cdp-dataflow-sa` (configurable) | Dedicated Dataflow worker service account with least-privilege roles (`roles/storage.objectAdmin`, `roles/dataflow.worker`, `roles/monitoring.metricWriter`, `roles/pubsub.editor`, `roles/bigquery.dataEditor`, `roles/bigquery.jobUser`). | -| Variable | Type | Description | -| :---------------------- | :-------: | :----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `billing_account` | `string` | Optional. Billing account to be used if the project is created from scratch. | -| `organization` | `string` | Optional. Organization (or folder) number, where the project will be created. Only required if you are creating the project from scratch. Use the format `organizations/XXXXXXX` or `folder/XXXXXXXX`. | -| `project_create` | `boolean` | Set to false to reuse an existing project. Or to true to create a new project from scratch. | -| `project_id` | `string` | Project Id. | -| `region` | `string` | Region to be used for all the resources. The VPC will contain only a single sub-network in this region. | -| `destroy_all_resources` | `bool` | Optional. Default true. Destroy buckets and the Spanner instance with `tf destroy `. | -| `network-prefix` | `string` | Optional. Default "dataflow". Add a prefix to the network net, subnet, NAT | -| `zone` | `string` | Zone to create bigtable clusters | +## Configuration variables -The default values of all the optional configuration variables are set for development projects. -**For a production project, you should change `destroy_all_resources` to false.** +| Variable | Type | Default | Description | +| :--- | :---: | :---: | :--- | +| `project_id` | `string` | *(Required)* | Project ID of the existing GCP project where resources will be provisioned. | +| `region` | `string` | *(Required)* | GCP region for application resources and Dataflow jobs (e.g. `us-central1`, `europe-west1`). | +| `subnetwork` | `string` | `null` | Optional subnetwork URL or path for Dataflow workers (e.g. `regions/europe-west1/subnetworks/dev-subnet` or full Shared VPC URI `https://www.googleapis.com/compute/v1/projects/HOST_PROJECT/regions/REGION/subnetworks/SUBNET_NAME`). If omitted, the default network is used. | +| `bucket_name` | `string` | `null` | Optional GCS bucket name for Dataflow temp/staging files. Defaults to `project_id`. | +| `service_account_name` | `string` | `"cdp-dataflow-sa"` | Dedicated Dataflow worker service account ID. | +| `artifact_registry_name` | `string` | `"cdp-containers"` | Name of the Artifact Registry repository for custom Dataflow worker containers. | +| `pubsub_transactions_topic` | `string` | `"cdp-transactions"` | Name of the Pub/Sub topic for streaming customer transaction events. | +| `pubsub_coupon_redemption_topic` | `string` | `"cdp-coupon-redemption"` | Name of the Pub/Sub topic for streaming coupon redemption events. | +| `create_bucket` | `bool` | `false` | Set to `true` to provision a new GCS bucket, or `false` to reuse an existing bucket. | +| `destroy_all_resources` | `bool` | `true` | When `true`, enables deletion of BigQuery dataset contents and tables on `terraform destroy`. Set to `false` for production environments. | +| `bq_dataset` | `string` | `"cdp_dataset"` | The BigQuery output dataset name for customer data unification. | +| `bq_table` | `string` | `"unified_customer_data"` | The BigQuery output table name for unified customer data. | ## How to deploy -1. **Set the configuration variables:** +1. **Set configuration variables:** - - Create a file named `terraform.tfvars` in the current directory. - - Add the following configuration variables to the file, replacing the values with your own: + Create a `terraform.tfvars` file in this directory. - ```bash - billing_account = "YOUR_BILLING_ACCOUNT" - organization = "YOUR_ORGANIZATION_ID" - project_create = TRUE_OR_FALSE - project_id = "YOUR_PROJECT_ID" - region = "YOUR_REGION" - ``` + **Standard Deployment (Default Network / Same Project):** + ```hcl + project_id = "YOUR_PROJECT_ID" + region = "us-central1" + destroy_all_resources = true + ``` - - If this is a production deployment, make sure you change also the optional variables. + **Shared VPC Deployment (Dataflow in Service Project, Network in Host Project):** + ```hcl + project_id = "YOUR_SERVICE_PROJECT_ID" + region = "europe-west1" + subnetwork = "https://www.googleapis.com/compute/v1/projects/HOST_PROJECT_ID/regions/europe-west1/subnetworks/shared-dataflow-subnet" + bucket_name = "YOUR_STAGING_BUCKET" + create_bucket = false + service_account_name = "cdp-dataflow-sa" + destroy_all_resources = true + ``` 2. **Initialize Terraform:** - - - Run the following command to initialize Terraform: - - ```bash - terraform init - ``` + ```bash + terraform init + ``` 3. **Apply the configuration:** - - - Run the following command to apply the Terraform configuration: - - ```bash - terraform apply - ``` - -4. **Wait for the deployment to complete:** - - Terraform will output the status of the deployment. Wait for it to complete successfully. -5. **Access the deployed resources:** You are now ready to launch the sample pipeline in this - solution guide. + ```bash + terraform plan -out=tfplan + terraform apply tfplan + ``` + +4. **Access the deployed resources:** + Terraform will automatically generate `pipelines/cdp/scripts/00_set_environment.sh` with all required environment variables. + + Proceed to the pipeline directory to build the container and launch the streaming pipeline: + ```bash + cd ../../pipelines/cdp + source scripts/00_set_environment.sh + ./scripts/01_build_and_push_container.sh + ./scripts/02_run_dataflow.sh + ``` ## How to remove -The setup will be continuously consuming as this is a streaming architecture, running without stop. - -**BEWARE: THE COMMAND BELOW WILL DESTROY AND REMOVE ALL THE RESOURCES**. +To destroy all provisioned infrastructure: -To destroy and stop all the resources, run: +1. Cancel any active Dataflow streaming jobs first: + ```bash + gcloud dataflow jobs list --region=YOUR_REGION --status=active + gcloud dataflow jobs cancel JOB_ID --region=YOUR_REGION + ``` +2. Run `terraform destroy`: + ```bash + terraform destroy + ``` -```bash -terraform destroy -``` diff --git a/terraform/cdp/main.tf b/terraform/cdp/main.tf index cb61c42b..a53a2520 100644 --- a/terraform/cdp/main.tf +++ b/terraform/cdp/main.tf @@ -13,41 +13,78 @@ # limitations under the License. locals { - dataflow_service_account = "my-dataflow-sa" + bucket_name = var.bucket_name != null ? var.bucket_name : var.project_id + dataflow_service_account = var.service_account_name != null ? var.service_account_name : "cdp-dataflow-sa" max_dataflow_workers = 1 worker_disk_size_gb = 200 machine_type = "e2-standard-8" + bigquery_dataset = var.bq_dataset + bigquery_table = var.bq_table + transactions_topic = var.pubsub_transactions_topic + transactions_sub = "${var.pubsub_transactions_topic}-sub" + coupon_redemption_topic = var.pubsub_coupon_redemption_topic + coupon_redemption_sub = "${var.pubsub_coupon_redemption_topic}-sub" + artifact_registry_repo = var.artifact_registry_name } +data "google_project" "project" { + project_id = var.project_id +} -// Project -module "google_cloud_project" { - source = "github.com/GoogleCloudPlatform/cloud-foundation-fabric//modules/project?ref=v58.0.0" - billing_account = var.billing_account - project_reuse = var.project_create ? null : {} - name = var.project_id - parent = var.organization - services = [ - "cloudbuild.googleapis.com", - "dataflow.googleapis.com", - "monitoring.googleapis.com", - "pubsub.googleapis.com", - "autoscaling.googleapis.com", - "artifactregistry.googleapis.com", - "bigquery.googleapis.com", - "sqladmin.googleapis.com", - ] +// Enable required Google Cloud APIs natively +resource "google_project_service" "dataflow" { + project = var.project_id + service = "dataflow.googleapis.com" + disable_on_destroy = false +} + +resource "google_project_service" "cloudbuild" { + project = var.project_id + service = "cloudbuild.googleapis.com" + disable_on_destroy = false +} + +resource "google_project_service" "artifactregistry" { + project = var.project_id + service = "artifactregistry.googleapis.com" + disable_on_destroy = false +} + +resource "google_project_service" "pubsub" { + project = var.project_id + service = "pubsub.googleapis.com" + disable_on_destroy = false +} + +resource "google_project_service" "bigquery" { + project = var.project_id + service = "bigquery.googleapis.com" + disable_on_destroy = false +} + +resource "google_project_service" "monitoring" { + project = var.project_id + service = "monitoring.googleapis.com" + disable_on_destroy = false } +resource "google_project_service" "storage" { + project = var.project_id + service = "storage.googleapis.com" + disable_on_destroy = false +} + +// Artifact Registry repository for custom Dataflow worker containers module "registry_docker" { source = "github.com/GoogleCloudPlatform/cloud-foundation-fabric//modules/artifact-registry?ref=v58.0.0" - project_id = module.google_cloud_project.project_id + project_id = var.project_id location = var.region - name = "dataflow-containers" + name = local.artifact_registry_repo format = { docker = { standard = {} } } iam = { "roles/artifactregistry.admin" = [ - "serviceAccount:${module.google_cloud_project.number}@cloudbuild.gserviceaccount.com" + "serviceAccount:${data.google_project.project.number}@cloudbuild.gserviceaccount.com", + "serviceAccount:${data.google_project.project.number}-compute@developer.gserviceaccount.com" ] "roles/artifactregistry.reader" = [ module.dataflow_sa.iam_email @@ -62,126 +99,135 @@ module "registry_docker" { } } } -} + depends_on = [ + google_project_service.artifactregistry + ] +} -// Buckets for staging data, scripts, etc, in the two regions +// Optional GCS Bucket for staging data and scripts module "buckets" { + count = var.create_bucket ? 1 : 0 source = "github.com/GoogleCloudPlatform/cloud-foundation-fabric//modules/gcs?ref=v58.0.0" - project_id = module.google_cloud_project.project_id - name = module.google_cloud_project.project_id + project_id = var.project_id + name = local.bucket_name location = var.region storage_class = "STANDARD" force_destroy = var.destroy_all_resources + + depends_on = [ + google_project_service.storage + ] } +// Pub/Sub transactions topic and subscription module "transactions_topic" { source = "github.com/GoogleCloudPlatform/cloud-foundation-fabric//modules/pubsub?ref=v58.0.0" - project_id = module.google_cloud_project.project_id - name = "transactions" + project_id = var.project_id + name = local.transactions_topic subscriptions = { - transactions-sub = {} + (local.transactions_sub) = {} } + + depends_on = [ + google_project_service.pubsub + ] } +// Pub/Sub coupon redemption topic and subscription module "coupon_redemption_topic" { source = "github.com/GoogleCloudPlatform/cloud-foundation-fabric//modules/pubsub?ref=v58.0.0" - project_id = module.google_cloud_project.project_id - name = "coupon_redemption" + project_id = var.project_id + name = local.coupon_redemption_topic subscriptions = { - coupon_redemption-sub = {} + (local.coupon_redemption_sub) = {} } + + depends_on = [ + google_project_service.pubsub + ] } -//bigquery dataset -module "output_dataset" { +// BigQuery dataset for unified CDP data +module "cdp_dataset" { source = "github.com/GoogleCloudPlatform/cloud-foundation-fabric//modules/bigquery-dataset?ref=v58.0.0" - project_id = module.google_cloud_project.project_id - id = var.bq_dataset + project_id = var.project_id + id = local.bigquery_dataset + location = var.region + access = { + dataflow-writer = { role = "OWNER", type = "user" } + } + access_identities = { + dataflow-writer = module.dataflow_sa.email + } + options = { + delete_contents_on_destroy = var.destroy_all_resources + } + + depends_on = [ + google_project_service.bigquery + ] } +// BigQuery destination table for unified customer transactions and coupons +resource "google_bigquery_table" "unified_customer_data" { + project = var.project_id + dataset_id = module.cdp_dataset.dataset_id + table_id = local.bigquery_table + deletion_protection = !var.destroy_all_resources + + schema = jsonencode([ + { name = "transaction_id", type = "STRING", mode = "REQUIRED" }, + { name = "household_key", type = "STRING", mode = "NULLABLE" }, + { name = "coupon_upc", type = "STRING", mode = "NULLABLE" }, + { name = "product_id", type = "STRING", mode = "NULLABLE" }, + { name = "coupon_discount", type = "STRING", mode = "NULLABLE" } + ]) + + depends_on = [ + module.cdp_dataset + ] +} -// Service account +// Dedicated Dataflow Worker Service Account module "dataflow_sa" { source = "github.com/GoogleCloudPlatform/cloud-foundation-fabric//modules/iam-service-account?ref=v58.0.0" - project_id = module.google_cloud_project.project_id + project_id = var.project_id name = local.dataflow_service_account iam_project_roles = { - (module.google_cloud_project.project_id) = [ - "roles/storage.admin", + (var.project_id) = [ + "roles/storage.objectAdmin", "roles/dataflow.worker", "roles/monitoring.metricWriter", "roles/pubsub.editor", - "roles/bigquery.dataEditor" + "roles/bigquery.dataEditor", + "roles/bigquery.jobUser" ] } } - -// Network -module "vpc_network" { - source = "github.com/GoogleCloudPlatform/cloud-foundation-fabric//modules/net-vpc?ref=v58.0.0" - project_id = module.google_cloud_project.project_id - name = "${var.network_prefix}-net" - subnets = [ - { - ip_cidr_range = "10.1.0.0/16" - name = "${var.network_prefix}-subnet" - region = var.region - enable_private_access = true - secondary_ip_ranges = { - pods = { ip_cidr_range = "10.16.0.0/14" } - services = { ip_cidr_range = "10.20.0.0/24" } - } - } - ] -} - -module "firewall_rules" { - // Default rules for internal traffic + SSH access via IAP - source = "github.com/GoogleCloudPlatform/cloud-foundation-fabric//modules/net-vpc-firewall?ref=v58.0.0" - project_id = module.google_cloud_project.project_id - network = module.vpc_network.name - default_rules_config = { - admin_ranges = [ - module.vpc_network.subnet_ips["${var.region}/${var.network_prefix}-subnet"], - ] - } - egress_rules = { - allow-egress-dataflow = { - deny = false - description = "Dataflow firewall rule egress" - targets = ["dataflow"] - rules = [{ protocol = "tcp", ports = [12345, 12346] }] - } - } - ingress_rules = { - allow-ingress-dataflow = { - description = "Dataflow firewall rule ingress" - targets = ["dataflow"] - rules = [{ protocol = "tcp", ports = [12345, 12346] }] - } - } -} -module "regional_nat" { - // So we can get to Internet if necessary (from the Dataflow region) - source = "github.com/GoogleCloudPlatform/cloud-foundation-fabric//modules/net-cloudnat?ref=v58.0.0" - project_id = module.google_cloud_project.project_id - region = var.region - name = "${var.network_prefix}-nat" - router_network = module.vpc_network.self_link +// Grant networkUser role on the subnetwork to the Dataflow worker service account (supports Shared VPC and local subnets) +resource "google_compute_subnetwork_iam_member" "dataflow_network_user" { + count = var.subnetwork != null ? 1 : 0 + project = length(regexall("projects/([^/]+)/", var.subnetwork)) > 0 ? regex("projects/([^/]+)/", var.subnetwork)[0] : var.project_id + region = length(regexall("regions/([^/]+)/", var.subnetwork)) > 0 ? regex("regions/([^/]+)/", var.subnetwork)[0] : var.region + subnetwork = length(regexall("subnetworks/([^/]+)", var.subnetwork)) > 0 ? regex("subnetworks/([^/]+)", var.subnetwork)[0] : var.subnetwork + role = "roles/compute.networkUser" + member = module.dataflow_sa.iam_email } +// Script with variables to launch the Dataflow jobs resource "local_file" "variables_script" { - filename = "${path.module}/../../pipelines/cdp/scripts/00_set_variables.sh" + filename = "${path.module}/../../pipelines/cdp/scripts/00_set_environment.sh" file_permission = "0644" content = <