diff --git a/.agents/skills/use-case-deployment/SKILL.md b/.agents/skills/use-case-deployment/SKILL.md index 8a2eb70f..d2df1847 100644 --- a/.agents/skills/use-case-deployment/SKILL.md +++ b/.agents/skills/use-case-deployment/SKILL.md @@ -102,11 +102,13 @@ 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 +# 3. Generate Streaming Transactions & Shopping Sessions +python3 ./cdp_pipeline/generate_transaction_data.py --continuous --interval=1.0 -# 4. Validate Unified BigQuery Table -bq query --use_legacy_sql=false 'SELECT * FROM cdp_dataset.unified_customer_data LIMIT 10' +# 4. Validate BigQuery Tables (Unified items, Customer 360 Sessions, Deadletter) +bq query --use_legacy_sql=false 'SELECT session_id, household_key, product_id, sales_value, coupon_upc FROM cdp_dataset.unified_customer_data LIMIT 10' +bq query --use_legacy_sql=false 'SELECT session_id, household_key, total_spend, total_items_purchased, coupons_redeemed_count FROM cdp_dataset.customer_sessions LIMIT 10' +bq query --use_legacy_sql=false 'SELECT * FROM cdp_dataset.cdp_deadletter LIMIT 10' ``` ### 4. Clickstream Analytics with Bigtable (Java) diff --git a/AGENTS.md b/AGENTS.md index 82b95f80..2373db18 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -179,7 +179,7 @@ When assisting a user with deploying a solution guide, follow this structured 7- ./scripts/02_run_dataflow.sh # (or ./scripts/01_launch_pipeline.sh) ``` 5. **Data Ingestion & Simulation**: - - Run the data generator or publisher script to produce streaming events (e.g. `python cdp_pipeline/generate_transaction_data.py` or publishing to Pub/Sub). + - Run the data generator or publisher script to produce streaming events (e.g. `python scripts/03_publish_events.py` or publishing to Pub/Sub). 6. **Verification & Observability**: - Inspect Dataflow Job status via GCP Console or `gcloud dataflow jobs list`. - Query target destinations (BigQuery tables, Cloud Spanner database, Cloud Bigtable rows, Pub/Sub output subscriptions). diff --git a/README.md b/README.md index c703784a..24a3a099 100644 --- a/README.md +++ b/README.md @@ -27,7 +27,7 @@ This the list of solution guides available at this moment: | [Clickstream Analytics](./use_cases/Clickstream_Analytics.md) | Real-time clickstream analytics with Bigtable enrichment / data hydration | Ready :white_check_mark: | | [IoT Analytics](./use_cases/IoT_Analytics.md) | Real-time Internet of Things (IoT) analytics with Bigtable enrichment & Scikit-Learn RunInference | Ready :white_check_mark: | | [Anomaly Detection](./use_cases/Anomaly_Detection.md) | Real-time anomaly detection with Bigtable enrichment & models deployed in Vertex AI | Ready :white_check_mark: | -| [Customer Data Platform](./use_cases/CDP.md) | Real-time customer data platform that unifies a customer view from different sources. | Beta :factory: | +| [Customer Data Platform](./use_cases/CDP.md) | Real-time customer data platform that unifies a customer view from different sources. | Ready :white_check_mark: | | [Gaming Analytics](./use_cases/gaming_analytics.md) | Real-time analyis of gaming data to enhance live gameplay & offer targeting | Beta :factory: | ## Repository structure diff --git a/pipelines/cdp/.dockerignore b/pipelines/cdp/.dockerignore new file mode 100644 index 00000000..ac382326 --- /dev/null +++ b/pipelines/cdp/.dockerignore @@ -0,0 +1,8 @@ +.venv/ +.pytest_cache/ +__pycache__/ +*.pyc +.git/ +dist/ +build/ +*.egg-info/ diff --git a/pipelines/cdp/.gcloudignore b/pipelines/cdp/.gcloudignore new file mode 100644 index 00000000..ac382326 --- /dev/null +++ b/pipelines/cdp/.gcloudignore @@ -0,0 +1,8 @@ +.venv/ +.pytest_cache/ +__pycache__/ +*.pyc +.git/ +dist/ +build/ +*.egg-info/ diff --git a/pipelines/cdp/Dockerfile b/pipelines/cdp/Dockerfile index 6671345b..c4e32e20 100644 --- a/pipelines/cdp/Dockerfile +++ b/pipelines/cdp/Dockerfile @@ -27,7 +27,9 @@ COPY setup.py setup.py RUN pip install --upgrade --no-cache-dir pip \ && pip install --no-cache-dir -r requirements.txt \ - && pip install --no-cache-dir -e . + && pip install --no-cache-dir . + +ENV PYTHONPATH="/workspace:${PYTHONPATH:-}" # Copy files from official SDK image, including script/dependencies. COPY --from=apache/beam_python3.14_sdk:2.76.0 /opt/apache/beam /opt/apache/beam diff --git a/pipelines/cdp/MANIFEST.in b/pipelines/cdp/MANIFEST.in index 540b7204..7c546978 100644 --- a/pipelines/cdp/MANIFEST.in +++ b/pipelines/cdp/MANIFEST.in @@ -1 +1,3 @@ -include requirements.txt \ No newline at end of file +include requirements.txt +include LICENSE +recursive-include schema *.json \ No newline at end of file diff --git a/pipelines/cdp/README.md b/pipelines/cdp/README.md index 8355668b..e5e86f24 100644 --- a/pipelines/cdp/README.md +++ b/pipelines/cdp/README.md @@ -1,85 +1,137 @@ -# Customer Data Platform sample pipeline (Python) +# Customer Data Platform Streaming Pipeline (Python) -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 production-grade streaming pipeline demonstrates how to use Google Cloud Dataflow and Apache Beam to implement an end-to-end Customer Data Platform (CDP). It ingests multi-stream customer interactions from Cloud Pub/Sub (`cdp-transactions` and `cdp-coupon-redemption`), reconstructs individual customer shopping sessions using dynamic **`Sessions(gap_size)` windowing**, unifies transaction baskets with redeemed coupons, aggregates **Customer 360 session profiles**, routes malformed inputs to a **Dead-Letter Queue (DLQ)**, and streams output records to **BigQuery** using the high-performance **Storage Write API**. This pipeline is part of the [Dataflow Customer Data Platform solution guide](../../use_cases/CDP.md). ## Architecture -The generic architecture for the CDP pipeline looks as follows: +The real-time sessionization architecture operates as follows: -![Architecture](../imgs/cdp.png) +1. **Multi-Stream Ingestion**: Reads streaming events from Pub/Sub subscriptions (`cdp-transactions-sub` and `cdp-coupon-redemption-sub`) with automatic topic fallback. +2. **Safe Deserialization & Dead-Letter Routing**: `ParseRecordDoFn` validates JSON payloads and verifies mandatory keys (`household_key`, `transaction_id`). Invalid payloads or schema violations are tagged as `errors` and routed to the dead-letter sink. +3. **Event-Time Timestamping & Sessionization**: Records are timestamped based on `event_timestamp` and windowed into dynamic session windows via `Sessions(gap_size)` (default 15 minutes). Watermark-based late data triggers and accumulating modes ensure late events are incorporated into sessions safely. +4. **Customer 360 Session Aggregation**: `ProcessCustomerSessionDoFn` groups interactions by `household_key`, emitting: + - **Granular Unified Items** (main output): Each purchased item unified with its session ID, price, quantity, store, and applied coupon/discount. + - **Customer 360 Session Profiles** (tagged output `sessions`): Rollup of total session spend, item count, discounts, distinct products, stores visited, and marketing campaigns engaged. +5. **Storage Write API Dual Sinks**: High-throughput direct ingestion into BigQuery tables with deadletter error routing. -In this directory, you will find a specific implementation of the above architecture with the following stages: +## BigQuery Data Schemas -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`. +The pipeline outputs into three BigQuery tables defined in `schema/`: -## Selecting the cloud region +- **`cdp_dataset.unified_customer_data`** (`schema/unified_table.json`): Granular item-level purchases enriched with session ID, store, retail discount, coupon UPC, and campaign. +- **`cdp_dataset.customer_sessions`** (`schema/customer_sessions.json`): Aggregated Customer 360 profile per shopping session (session duration, total spend, total items, coupon count, campaigns, visited stores). +- **`cdp_dataset.cdp_deadletter`** (`schema/deadletter_table.json`): Error records, including original payload, error reason, source topic, and timestamp. -Not all resources may be available in all regions. The default values included in this directory have been tested using `us-central1` as region. +## How to Launch the Pipeline -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: +All launch scripts are located in the `scripts/` directory. -```sh -gcloud compute machine-types list --zones=,,... -``` - -See more info about selecting the right type of machine in Google Cloud Compute Engine documentation: -* https://cloud.google.com/compute/docs/machine-resource +### 1. Load Environment Variables +The environment configuration file `scripts/00_set_environment.sh` is generated automatically when deploying the Terraform infrastructure in `terraform/cdp/`: -## How to launch the pipeline - -All scripts are located in the `scripts` directory and prepared to be launched from the `pipelines/cdp` directory. +```bash +source scripts/00_set_environment.sh +``` -### 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: +### 2. Run Locally with DirectRunner (Optional for Development) +To test pipeline transforms locally with DirectRunner: -```sh -source scripts/00_set_environment.sh +```bash +./scripts/02_run_local.sh ``` -### 2. Build and publish custom container +### 3. Build and Publish Custom Container Build and push the custom Dataflow worker container to Artifact Registry using Cloud Build: -```sh +```bash ./scripts/01_build_and_push_container.sh ``` -### 3. Launch Dataflow streaming pipeline +### 4. Launch Dataflow Streaming Pipeline Submit the streaming pipeline job to Google Cloud Dataflow: -```sh +```bash ./scripts/02_run_dataflow.sh ``` -## Automated Tests +## Automated Tests & Code Quality -Execute unit and pipeline transform tests with `pytest`: +Execute unit, DoFn, and end-to-end pipeline transform tests with `pytest`: ```bash pytest tests/ -v ``` -## Input data simulation +Run code formatting and PyLint checks against Google Python style: -To send test data into the pipeline, publish messages to the `cdp-transactions` and `cdp-coupon-redemption` Pub/Sub topics: - -```python3 -python3 ./cdp_pipeline/generate_transaction_data.py +```bash +yapf -i -r --style yapf cdp_pipeline simulator scripts tests main.py +pylint --rcfile ../pylintrc cdp_pipeline simulator scripts/03_publish_events.py tests main.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. +## Input Data Simulation -## Output data +To publish streaming transactions and session journeys into Pub/Sub: -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 +```bash +# Continuous streaming mode (1 session journey per second) +python3 ./scripts/03_publish_events.py --continuous --interval=1.0 + +# Batch burst mode (100 sessions) +python3 ./scripts/03_publish_events.py --count=100 + +# Continuous mode with injected error payloads to test the DLQ +python3 ./scripts/03_publish_events.py --continuous --inject_errors ``` +## Output Data Verification + Verify output records via `bq`: + +```bash +# Inspect unified basket items +bq query --use_legacy_sql=false \ + "SELECT session_id, household_key, transaction_id, product_id, sales_value, coupon_upc \ + FROM \`${PROJECT}.cdp_dataset.unified_customer_data\` LIMIT 10" + +# Inspect Customer 360 session rollups +bq query --use_legacy_sql=false \ + "SELECT session_id, household_key, total_spend, total_items_purchased, coupons_redeemed_count \ + FROM \`${PROJECT}.cdp_dataset.customer_sessions\` LIMIT 10" + +# Inspect Dead-Letter Queue +bq query --use_legacy_sql=false \ + "SELECT source, error_message, timestamp \ + FROM \`${PROJECT}.cdp_dataset.cdp_deadletter\` LIMIT 10" +``` + +## Handling Late Data & BigQuery Deduplication + +The pipeline uses `AccumulationMode.ACCUMULATING` with an `allowed_lateness` window (default 60 seconds). When an out-of-order event (such as a late coupon redemption) arrives after the session watermark has closed, Dataflow re-evaluates the session: +1. It joins the late coupon with the earlier transactions in the session. +2. It recalculates the updated cumulative `CustomerSessionProfile`. +3. It appends the new records to BigQuery using `STORAGE_WRITE_API` (`WRITE_APPEND`). + +Because records are appended, late firings create updated versions of rows with a newer `processed_timestamp`. Downstream analytics and reporting views can deduplicate to retrieve the latest state using GoogleSQL's `QUALIFY` clause: + ```bash -bq query --use_legacy_sql=false "SELECT * FROM \`${PROJECT}.cdp_dataset.unified_customer_data\` LIMIT 10" -``` \ No newline at end of file +# Query latest Customer 360 session profiles +bq query --use_legacy_sql=false \ + "SELECT * \ + FROM \`${PROJECT}.cdp_dataset.customer_sessions\` \ + QUALIFY ROW_NUMBER() OVER ( \ + PARTITION BY session_id \ + ORDER BY processed_timestamp DESC \ + ) = 1 LIMIT 10" + +# Query latest unified basket items +bq query --use_legacy_sql=false \ + "SELECT * \ + FROM \`${PROJECT}.cdp_dataset.unified_customer_data\` \ + QUALIFY ROW_NUMBER() OVER ( \ + PARTITION BY transaction_id, product_id, COALESCE(coupon_upc, '') \ + ORDER BY processed_timestamp DESC \ + ) = 1 LIMIT 10" +``` diff --git a/pipelines/cdp/cdp_pipeline/customer_data_platform.py b/pipelines/cdp/cdp_pipeline/customer_data_platform.py deleted file mode 100644 index bffa900d..00000000 --- a/pipelines/cdp/cdp_pipeline/customer_data_platform.py +++ /dev/null @@ -1,180 +0,0 @@ -# Copyright 2025 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. -""" -Customer Data Platform analytics pipeline for the Dataflow Solution Guides. -""" - -import json -import logging -import os -from typing import Any, Generator, Iterable, Optional, Union - -import apache_beam as beam -from apache_beam import Pipeline, PCollection -from apache_beam.io.gcp.bigquery import WriteToBigQuery -from apache_beam.transforms.trigger import AccumulationMode, AfterProcessingTime, AfterWatermark -from apache_beam.transforms.window import FixedWindows -from cdp_pipeline.options import MyPipelineOptions - -DEFAULT_OUTPUT_SCHEMA: dict[str, Any] = { - "fields": [ - { - "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" - }, - ] -} - - -def load_output_schema( - schema_path: Optional[str] = None) -> Union[dict[str, Any], str]: - """Loads the BigQuery output table schema from a specified path or default package location.""" - if schema_path: - with open(schema_path, encoding="utf-8") as schema_file: - return json.load(schema_file) - - # Check relative to the schema directory inside the cdp pipeline package - default_schema_file = os.path.join( - os.path.dirname(os.path.dirname(__file__)), "schema", - "unified_table.json") - if os.path.exists(default_schema_file): - with open(default_schema_file, encoding="utf-8") as schema_file: - return json.load(schema_file) - - return DEFAULT_OUTPUT_SCHEMA - - -def left_join( - key_value_pair: tuple[Any, tuple[Iterable[dict[str, Any]], - Iterable[Optional[dict[str, Any]]]]] -) -> Generator[dict[str, Any], None, None]: - """Performs a left join between transaction and coupon redemption records.""" - _, values = key_value_pair - trans_values, coupon_redempt_values = values - coupon_list = list(coupon_redempt_values) - if not coupon_list: - coupon_list = [None] # Fill missing values with None - for trans_value in trans_values: - if trans_value is not None: - for coupon_redempt_value in coupon_list: - coupon_upc = None - if isinstance(coupon_redempt_value, dict): - raw_upc = coupon_redempt_value.get("coupon_upc") - if raw_upc is not None: - coupon_upc = str(raw_upc) - unified_data = { - "transaction_id": - str(trans_value["transaction_id"]), - "household_key": - str(trans_value["household_key"]), - "coupon_upc": - coupon_upc, - "product_id": - str(trans_value["product_id"]), - "coupon_discount": - str( - trans_value.get("coupon_disc", - trans_value.get("coupon_discount", "0"))), - } - yield unified_data - - -@beam.ptransform_fn -def _read_pub_sub_topic(p: Pipeline, topic: str) -> PCollection[str]: - msgs: PCollection[bytes] = ( - p - | "Read subscription" >> beam.io.ReadFromPubSub(topic=topic) - | "Decode Transactions" >> - beam.Map(lambda msg: json.loads(msg.decode("utf-8"))) - | "Add Transaction Key" >> beam.Map(lambda transaction: ((transaction[ - "transaction_id"], transaction["household_key"]), transaction)) - | "Window Transactions" >> beam.WindowInto( - FixedWindows(60), - trigger=AfterWatermark(early=AfterProcessingTime(10)), - accumulation_mode=AccumulationMode.DISCARDING)) - - return msgs - - -@beam.ptransform_fn -def _unify_data(pcolls: tuple[PCollection, PCollection]) -> PCollection[str]: - transactions_pcoll, coupons_redempt_pcoll = pcolls - unified_data = ((transactions_pcoll, coupons_redempt_pcoll) - | "Combine Transactions and Coupons" >> beam.CoGroupByKey() - | beam.FlatMap(left_join)) - return unified_data - - -@beam.ptransform_fn -def _write_to_bq(unified_pcoll: PCollection, project_id: str, - output_dataset: str, output_table: str, - unified_schema: Union[dict[str, Any], str]): - unified_pcoll | "Write to bigquery" >> \ - WriteToBigQuery( - project=project_id, - dataset=output_dataset, - table=output_table, - schema=unified_schema, - create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED, - write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND - ) - - -def create_and_run_pipeline(pipeline_options: MyPipelineOptions, - output_schema: Optional[Union[dict[str, Any], - str]] = None): - logging.info(pipeline_options) - - if output_schema is None: - schema_path = getattr(pipeline_options, "output_schema_path", None) - output_schema = load_output_schema(schema_path) - - with Pipeline(options=pipeline_options) as p: - - # Read transcation pub-sub topic - transactions_pcoll = p | "Read transactions topic" >> _read_pub_sub_topic( - topic=pipeline_options.transactions_topic) - # Read coupon_redemption pub-sub topic - coupons_redempt_pcoll = p | "Read coupon redemption topic" >> _read_pub_sub_topic( - topic=pipeline_options.coupons_redemption_topic) - - # call _unify_data to unify the data from two streaming sources - unified_data: PCollection = (transactions_pcoll, coupons_redempt_pcoll - ) | "Transform" >> _unify_data() - - # Write it to bigquery. Provide schema of the output table as parameter output_schema - unified_data | "Write to bigquery" >> _write_to_bq( - pipeline_options.project_id, pipeline_options.output_dataset, - pipeline_options.output_table, output_schema) diff --git a/pipelines/cdp/cdp_pipeline/generate_transaction_data.py b/pipelines/cdp/cdp_pipeline/generate_transaction_data.py deleted file mode 100644 index 7e800d1c..00000000 --- a/pipelines/cdp/cdp_pipeline/generate_transaction_data.py +++ /dev/null @@ -1,142 +0,0 @@ -# Copyright 2025 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. -""" -A data generator for the Customer Data Platform analytics pipeline. -""" - -import argparse -import asyncio -import json -import os -from google.cloud import pubsub_v1 -import pandas as pd - - -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] - - sample_transactions_id = [ - "27601281299", "27757099033", "28235291311", "27021203242", - "27101290145", "27853175697" - ] - - # 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) - - 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)) - - filtered_trans_df = transactions_df[transactions_df["transaction_id"].isin( - sample_transactions_id)] - filtered_coupons_df = coupons_df[coupons_df["transaction_id"].isin( - 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, - transactions_topic_path)) - - -async def publish_coupons(filtered_coupons_df, publisher, coupons_topic_path): - for _, row in filtered_coupons_df.iterrows(): - coupon_message = json.dumps(row.to_dict()).encode("utf-8") - print(coupon_message) - future = publisher.publish(coupons_topic_path, coupon_message) - print(f"Published coupon message ID: {future.result()}") - await asyncio.sleep(3) - - -async def publish_transactions(filtered_trans_df, publisher, - transactions_topic_path): - for _, row in filtered_trans_df.iterrows(): - transaction_message = json.dumps(row.to_dict()).encode("utf-8") - print(transaction_message) - future = publisher.publish(transactions_topic_path, transaction_message) - print(f"Published transaction message ID: {future.result()}") - await asyncio.sleep(1) - - -if __name__ == "__main__": - 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/cdp_pipeline/models.py b/pipelines/cdp/cdp_pipeline/models.py new file mode 100644 index 00000000..586259eb --- /dev/null +++ b/pipelines/cdp/cdp_pipeline/models.py @@ -0,0 +1,238 @@ +# 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. +"""Data models and Beam schemas for the Customer Data Platform pipeline.""" + +from datetime import datetime, timezone +from enum import StrEnum +import json +from typing import Any, Dict, List, NamedTuple, Optional, Tuple, Union + + +class EventType(StrEnum): + """Enumerates customer interaction event types.""" + TRANSACTION = "transaction" + COUPON = "coupon" + + +class TransactionItem(NamedTuple): + """Granular basket item in a customer transaction.""" + product_id: Optional[str] = None + quantity: int = 1 + sales_value: float = 0.0 + store_id: Optional[str] = None + retail_disc: float = 0.0 + coupon_disc: float = 0.0 + coupon_match_disc: float = 0.0 + day: Optional[int] = None + week_no: Optional[int] = None + trans_time: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + """Returns a dictionary representation of the transaction item.""" + return self._asdict() + + +class CouponRedemption(NamedTuple): + """Coupon redemption attached to a customer transaction.""" + coupon_upc: Optional[str] = None + campaign: Optional[str] = None + day: Optional[int] = None + + def to_dict(self) -> Dict[str, Any]: + """Returns a dictionary representation of the coupon redemption.""" + return self._asdict() + + +class DeadLetterRecord(NamedTuple): + """Represents a malformed or invalid payload routed to the Dead-Letter Queue.""" + source: str + raw_payload: str + error_message: str + timestamp: str + + def to_dict(self) -> Dict[str, Any]: + """Returns a dictionary representation suitable for BigQuery insertion.""" + return self._asdict() + + +class CustomerInteractionEvent(NamedTuple): + """Unified customer event composed of transaction or coupon data for sessionization.""" + event_type: str # EventType.TRANSACTION or EventType.COUPON + household_key: str + transaction_id: str + event_timestamp: Optional[str] = None + transaction: Optional[TransactionItem] = None + coupon: Optional[CouponRedemption] = None + + def to_dict(self) -> Dict[str, Any]: + """Returns a dictionary representation of the event.""" + return { + "event_type": self.event_type, + "household_key": self.household_key, + "transaction_id": self.transaction_id, + "event_timestamp": self.event_timestamp, + "transaction": self.transaction.to_dict() if self.transaction else None, + "coupon": self.coupon.to_dict() if self.coupon else None, + } + + @classmethod + def from_raw_payload( + cls, + element: Union[bytes, str, Dict[str, Any]], + expected_type: EventType, + ) -> Tuple[Optional["CustomerInteractionEvent"], Optional[DeadLetterRecord]]: + """Decodes and validates raw bytes/str/dict into a typed CustomerInteractionEvent.""" + try: + if isinstance(element, bytes): + payload_str = element.decode("utf-8") + data = json.loads(payload_str) + elif isinstance(element, str): + payload_str = element + data = json.loads(payload_str) + elif isinstance(element, dict): + payload_str = json.dumps(element) + data = dict(element) + else: + raise ValueError(f"Unsupported payload type: {type(element)}") + except Exception as exc: # pylint: disable=broad-exception-caught + now_iso = datetime.now(timezone.utc).isoformat() + return None, DeadLetterRecord( + source=expected_type.value, + raw_payload=str(element)[:2000], + error_message=f"Malformed payload: {exc}", + timestamp=now_iso, + ) + + hh_key = str(data.get("household_key", "")).strip() + tx_id = str(data.get("transaction_id", "")).strip() + if not hh_key or not tx_id: + now_iso = datetime.now(timezone.utc).isoformat() + return None, DeadLetterRecord( + source=expected_type.value, + raw_payload=payload_str[:2000], + error_message="Missing required household_key or transaction_id", + timestamp=now_iso, + ) + + event_ts = data.get("event_timestamp") + + if expected_type == EventType.TRANSACTION: + try: + qty = int(float(data.get("quantity", 1) or 1)) + except (ValueError, TypeError): + qty = 1 + try: + sales = float(data.get("sales_value", 0.0) or 0.0) + except (ValueError, TypeError): + sales = 0.0 + try: + ret_disc = float(data.get("retail_disc", 0.0) or 0.0) + except (ValueError, TypeError): + ret_disc = 0.0 + try: + coup_disc = float( + data.get("coupon_disc", data.get("coupon_discount", 0.0)) or 0.0) + except (ValueError, TypeError): + coup_disc = 0.0 + try: + match_disc = float(data.get("coupon_match_disc", 0.0) or 0.0) + except (ValueError, TypeError): + match_disc = 0.0 + + day_val = data.get("day") + week_val = data.get("week_no") + + tx_item = TransactionItem( + product_id=str(data.get("product_id", "")).strip() or None, + quantity=qty, + sales_value=sales, + store_id=str(data.get("store_id", "")).strip() or None, + retail_disc=ret_disc, + coupon_disc=coup_disc, + coupon_match_disc=match_disc, + day=int(day_val) if day_val is not None else None, + week_no=int(week_val) if week_val is not None else None, + trans_time=str(data.get("trans_time", "")).strip() or None, + ) + return cls( + event_type=EventType.TRANSACTION.value, + household_key=hh_key, + transaction_id=tx_id, + event_timestamp=event_ts, + transaction=tx_item, + coupon=None, + ), None + + # Coupon event + day_val = data.get("day") + coupon_item = CouponRedemption( + coupon_upc=str(data.get("coupon_upc", "")).strip() or None, + campaign=str(data.get("campaign", "")).strip() or None, + day=int(day_val) if day_val is not None else None, + ) + return cls( + event_type=EventType.COUPON.value, + household_key=hh_key, + transaction_id=tx_id, + event_timestamp=event_ts, + transaction=None, + coupon=coupon_item, + ), None + + +class UnifiedTransactionRecord(NamedTuple): + """Granular customer transaction item enriched with session ID and discounts.""" + session_id: str + transaction_id: str + household_key: str + product_id: Optional[str] + quantity: Optional[int] + sales_value: Optional[float] + store_id: Optional[str] + retail_disc: Optional[float] + coupon_discount: Optional[float] + coupon_match_disc: Optional[float] + coupon_upc: Optional[str] + campaign: Optional[str] + day: Optional[int] + trans_time: Optional[str] + week_no: Optional[int] + event_timestamp: Optional[str] + processed_timestamp: str + + def to_dict(self) -> Dict[str, Any]: + """Returns a dictionary representation suitable for BigQuery insertion.""" + return self._asdict() + + +class CustomerSessionProfile(NamedTuple): + """Aggregated Customer 360 session profile.""" + session_id: str + household_key: str + session_start: str + session_end: str + session_duration_sec: int + total_transactions: int + total_items_purchased: int + total_spend: float + total_discount: float + coupons_redeemed_count: int + distinct_products_count: int + campaigns: List[str] + stores_visited: List[str] + processed_timestamp: str + + def to_dict(self) -> Dict[str, Any]: + """Returns a dictionary representation suitable for BigQuery insertion.""" + return self._asdict() diff --git a/pipelines/cdp/cdp_pipeline/options.py b/pipelines/cdp/cdp_pipeline/options.py index b2537802..205dd4d3 100644 --- a/pipelines/cdp/cdp_pipeline/options.py +++ b/pipelines/cdp/cdp_pipeline/options.py @@ -1,4 +1,4 @@ -# Copyright 2025 Google LLC +# 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. @@ -11,9 +11,7 @@ # 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. -""" -Option class for Customer Data Platform pipeline. -""" +"""Option class for Customer Data Platform pipeline.""" from argparse import ArgumentParser @@ -21,18 +19,91 @@ class MyPipelineOptions(PipelineOptions): + """Pipeline options for Customer Data Platform streaming sessionization.""" @classmethod def _add_argparse_args(cls, parser: ArgumentParser): - parser.add_argument("--transactions_topic", type=str) - parser.add_argument("--coupons_redemption_topic", type=str) - parser.add_argument("--project_id", type=str) - parser.add_argument("--location", type=str) - parser.add_argument("--output_dataset", type=str) - parser.add_argument("--output_table", type=str) + parser.add_argument( + "--transactions_topic", + type=str, + default=None, + help="Pub/Sub topic for streaming customer transaction events.", + ) + parser.add_argument( + "--transactions_subscription", + type=str, + default=None, + help="Pub/Sub subscription for streaming customer transactions.", + ) + parser.add_argument( + "--coupons_redemption_topic", + type=str, + default=None, + help="Pub/Sub topic for streaming coupon redemption events.", + ) + parser.add_argument( + "--coupons_redemption_subscription", + type=str, + default=None, + help="Pub/Sub subscription for streaming coupon redemptions.", + ) + parser.add_argument( + "--output_dataset", + type=str, + default="cdp_dataset", + help="Destination BigQuery dataset.", + ) + parser.add_argument( + "--output_table", + type=str, + default="unified_customer_data", + help="Destination BigQuery table for granular unified transactions.", + ) + parser.add_argument( + "--output_sessions_table", + type=str, + default="customer_sessions", + help="Destination BigQuery table for sessionized Customer 360 profiles.", + ) + parser.add_argument( + "--deadletter_table", + type=str, + default=None, + help="Optional BigQuery table name for rejected or malformed records.", + ) + parser.add_argument( + "--session_gap_seconds", + type=int, + default=900, + help="Session inactivity gap in seconds (default: 900s / 15m).", + ) + parser.add_argument( + "--allowed_lateness_seconds", + type=int, + default=60, + help="Allowed lateness in seconds for late-arriving events.", + ) + parser.add_argument( + "--use_storage_write_api", + action="store_true", + default=True, + help="Use BigQuery Storage Write API for high-throughput streaming.", + ) parser.add_argument( "--output_schema_path", type=str, default=None, - help="Optional path to custom JSON schema file for BigQuery output table.", + help="Optional path to custom JSON schema file for unified output table.", + ) + parser.add_argument( + "--output_sessions_schema_path", + type=str, + default=None, + help="Optional path to custom JSON schema file for session profiles.", + ) + parser.add_argument( + "--deadletter_schema_path", + type=str, + default=None, + help="Optional path to custom JSON schema file for deadletter table.", ) diff --git a/pipelines/cdp/cdp_pipeline/parsing.py b/pipelines/cdp/cdp_pipeline/parsing.py new file mode 100644 index 00000000..f5e45210 --- /dev/null +++ b/pipelines/cdp/cdp_pipeline/parsing.py @@ -0,0 +1,90 @@ +# 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. +"""Record parsing, payload validation, and timestamp assignment transforms.""" + +from datetime import datetime, timezone +from typing import Any, Dict, Generator, Tuple, Union + +import apache_beam as beam +from apache_beam.metrics import Metrics +from apache_beam.transforms.window import TimestampedValue + +from cdp_pipeline.models import CustomerInteractionEvent, EventType + +TAG_DEADLETTER = "errors" + + +class ParseRecordDoFn(beam.DoFn): + """Safely decodes and validates incoming JSON messages with Dead-Letter side outputs.""" + + def __init__(self, record_type: Union[EventType, str]): + super().__init__() + if isinstance(record_type, str): + self.record_type = EventType(record_type) + else: + self.record_type = record_type + self.processed_counter = None + self.error_counter = None + + def setup(self): + self.processed_counter = Metrics.counter( + self.__class__, f"processed_{self.record_type.value}") + self.error_counter = Metrics.counter(self.__class__, + f"error_{self.record_type.value}") + + def process( + self, + element: Union[bytes, str, Dict[str, Any]], + ) -> Generator[Any, None, None]: + event, deadletter = CustomerInteractionEvent.from_raw_payload( + element, self.record_type) + if deadletter is not None: + self.error_counter.inc() + yield beam.pvalue.TaggedOutput(TAG_DEADLETTER, deadletter.to_dict()) + return + + self.processed_counter.inc() + yield (event.household_key, event) + + +class AssignEventTimestampDoFn(beam.DoFn): + """Assigns event timestamp for session windowing based on payload event_timestamp.""" + + def process( + self, + element: Tuple[str, CustomerInteractionEvent], + timestamp=beam.DoFn.TimestampParam, + ) -> Generator[Any, None, None]: + _, event = element + event_ts_str = event.event_timestamp + ts_seconds = None + if event_ts_str: + try: + dt = datetime.fromisoformat(event_ts_str.replace("Z", "+00:00")) + ts_seconds = dt.timestamp() + except (ValueError, TypeError): + ts_seconds = None + + if ts_seconds is None: + try: + current_micros = timestamp.micros + if current_micros > 0: + ts_seconds = current_micros / 1000000.0 + except (AttributeError, TypeError, ValueError): + pass + + if ts_seconds is None or ts_seconds <= 0: + ts_seconds = datetime.now(timezone.utc).timestamp() + + yield TimestampedValue(element, ts_seconds) diff --git a/pipelines/cdp/cdp_pipeline/pipeline.py b/pipelines/cdp/cdp_pipeline/pipeline.py new file mode 100644 index 00000000..38be365a --- /dev/null +++ b/pipelines/cdp/cdp_pipeline/pipeline.py @@ -0,0 +1,142 @@ +# 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. +"""Customer Data Platform analytics and sessionization streaming pipeline.""" + +import logging +from typing import Any, Iterable, Optional, Tuple + +import apache_beam as beam +from apache_beam import Pipeline, PCollection +from apache_beam.transforms.trigger import AccumulationMode, AfterCount, AfterWatermark +from apache_beam.transforms.window import Sessions +from apache_beam.utils.timestamp import Duration + +from cdp_pipeline.models import EventType +from cdp_pipeline.options import MyPipelineOptions +from cdp_pipeline.parsing import ( + AssignEventTimestampDoFn, + ParseRecordDoFn, + TAG_DEADLETTER, +) +from cdp_pipeline.sessionization import ( + ProcessCustomerSessionDoFn, + TAG_SESSIONS, +) +from cdp_pipeline.sinks import apply_bigquery_sinks + + +def build_pipeline( + pipeline: Pipeline, + pipeline_options: MyPipelineOptions, + in_memory_transactions: Optional[Iterable[Any]] = None, + in_memory_coupons: Optional[Iterable[Any]] = None, +) -> Tuple[PCollection, PCollection, PCollection]: + """Builds the streaming sessionization pipeline graph on the given Pipeline object.""" + # 1. Read transactions stream + if in_memory_transactions is not None: + raw_transactions = pipeline | "Create Transactions" >> beam.Create( + in_memory_transactions) + elif pipeline_options.transactions_subscription: + raw_transactions = pipeline | "Read Transactions Sub" >> beam.io.ReadFromPubSub( + subscription=pipeline_options.transactions_subscription) + elif pipeline_options.transactions_topic: + raw_transactions = pipeline | "Read Transactions Topic" >> beam.io.ReadFromPubSub( + topic=pipeline_options.transactions_topic) + else: + raw_transactions = pipeline | "Empty Transactions" >> beam.Create([]) + + # 2. Read coupons stream + if in_memory_coupons is not None: + raw_coupons = pipeline | "Create Coupons" >> beam.Create(in_memory_coupons) + elif pipeline_options.coupons_redemption_subscription: + raw_coupons = pipeline | "Read Coupons Sub" >> beam.io.ReadFromPubSub( + subscription=pipeline_options.coupons_redemption_subscription) + elif pipeline_options.coupons_redemption_topic: + raw_coupons = pipeline | "Read Coupons Topic" >> beam.io.ReadFromPubSub( + topic=pipeline_options.coupons_redemption_topic) + else: + raw_coupons = pipeline | "Empty Coupons" >> beam.Create([]) + + # 3. Parse and extract customer key with dead-letter side outputs + parsed_tx_results = ( + raw_transactions + | "Parse Transactions" >> beam.ParDo( + ParseRecordDoFn(EventType.TRANSACTION)).with_outputs( + TAG_DEADLETTER, main="valid")) + + parsed_coupon_results = ( + raw_coupons + | "Parse Coupons" >> beam.ParDo(ParseRecordDoFn( + EventType.COUPON)).with_outputs(TAG_DEADLETTER, main="valid")) + + valid_transactions = parsed_tx_results.valid + valid_coupons = parsed_coupon_results.valid + + all_deadletters = ( + (parsed_tx_results[TAG_DEADLETTER], parsed_coupon_results[TAG_DEADLETTER]) + | "Merge Deadletter Errors" >> beam.Flatten()) + + # 4. Merge valid streams and apply dynamic Session Windows + gap_sec = getattr(pipeline_options, "session_gap_seconds", 900) or 900 + allowed_lateness_sec = getattr(pipeline_options, "allowed_lateness_seconds", + 60) or 60 + + trigger = ( + AfterWatermark( + late=AfterCount(1)) if allowed_lateness_sec > 0 else AfterWatermark()) + + # Note: AccumulationMode.ACCUMULATING ensures late-arriving events (such as + # delayed coupon redemptions) can still join against prior transactions in the + # session and recalculate complete session aggregates. Because BigQuery sinks use + # WRITE_APPEND, late panes append updated records which can be deduplicated + # downstream in BigQuery views using processed_timestamp. + all_events = ((valid_transactions, valid_coupons) + | "Merge Customer Streams" >> beam.Flatten() + | "Assign Timestamps" >> beam.ParDo(AssignEventTimestampDoFn()) + | "Customer Session Window" >> beam.WindowInto( + Sessions(gap_sec), + allowed_lateness=Duration(seconds=allowed_lateness_sec), + trigger=trigger, + accumulation_mode=AccumulationMode.ACCUMULATING, + ) + | "Group Customer Events" >> beam.GroupByKey()) + + # 5. Process Sessions to produce granular unified records and Customer 360 profiles + session_results = ( + all_events + | "Process Customer Sessions" >> beam.ParDo( + ProcessCustomerSessionDoFn()).with_outputs( + TAG_SESSIONS, main="unified_records")) + + unified_records = session_results.unified_records + customer_sessions = session_results[TAG_SESSIONS] + + # 6. Apply Storage Write API Sinks + apply_bigquery_sinks( + unified_records=unified_records, + customer_sessions=customer_sessions, + all_deadletters=all_deadletters, + pipeline_options=pipeline_options, + ) + + return unified_records, customer_sessions, all_deadletters + + +def create_and_run_pipeline(pipeline_options: MyPipelineOptions): + """Launches the Customer Data Platform streaming pipeline on Dataflow or DirectRunner.""" + logging.info("Starting Customer Data Platform pipeline with options: %s", + pipeline_options) + + with Pipeline(options=pipeline_options) as p: + build_pipeline(p, pipeline_options) diff --git a/pipelines/cdp/cdp_pipeline/schemas.py b/pipelines/cdp/cdp_pipeline/schemas.py new file mode 100644 index 00000000..e18995ae --- /dev/null +++ b/pipelines/cdp/cdp_pipeline/schemas.py @@ -0,0 +1,38 @@ +# 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. +"""BigQuery table schema loader for Customer Data Platform.""" + +import json +import os +from typing import Any, Dict, Optional, Union + + +def load_output_schema( + schema_path: Optional[str] = None, + default_filename: str = "unified_table.json", +) -> Union[Dict[str, Any], str]: + """Loads a BigQuery schema from a custom path or packaged schema JSON file.""" + if schema_path: + with open(schema_path, encoding="utf-8") as schema_file: + return json.load(schema_file) + + # Check package schema directory + default_schema_file = os.path.join( + os.path.dirname(os.path.dirname(__file__)), "schema", default_filename) + if os.path.exists(default_schema_file): + with open(default_schema_file, encoding="utf-8") as schema_file: + return json.load(schema_file) + + raise FileNotFoundError( + f"Schema file '{default_filename}' not found at {default_schema_file}.") diff --git a/pipelines/cdp/cdp_pipeline/sessionization.py b/pipelines/cdp/cdp_pipeline/sessionization.py new file mode 100644 index 00000000..883e3536 --- /dev/null +++ b/pipelines/cdp/cdp_pipeline/sessionization.py @@ -0,0 +1,171 @@ +# 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. +"""Session aggregation and Customer 360 profile generation transforms.""" + +import collections +from datetime import datetime, timezone +from typing import Any, Dict, Generator, Iterable, List, Optional, Tuple + +import apache_beam as beam +from apache_beam.metrics import Metrics + +from cdp_pipeline.models import ( + CouponRedemption, + CustomerInteractionEvent, + CustomerSessionProfile, + TransactionItem, + UnifiedTransactionRecord, +) + +TAG_SESSIONS = "sessions" + + +class ProcessCustomerSessionDoFn(beam.DoFn): + """Aggregates customer interactions into unified records and Customer 360 profiles.""" + + def __init__(self): + super().__init__() + self.sessions_counter = None + self.unified_items_counter = None + self.matched_coupons_counter = None + + def setup(self): + self.sessions_counter = Metrics.counter(self.__class__, + "completed_sessions") + self.unified_items_counter = Metrics.counter(self.__class__, + "unified_items_emitted") + self.matched_coupons_counter = Metrics.counter(self.__class__, + "matched_coupons") + + def process( + self, + element: Tuple[str, Iterable[CustomerInteractionEvent]], + window=beam.DoFn.WindowParam, + ) -> Generator[Any, None, None]: + household_key, events_iter = element + events = list(events_iter) + if not events: + return + + try: + session_start_iso = window.start.to_utc_datetime().isoformat() + except (OverflowError, ValueError): + session_start_iso = datetime.now(timezone.utc).isoformat() + + try: + session_end_iso = window.end.to_utc_datetime().isoformat() + except (OverflowError, ValueError): + session_end_iso = datetime.now(timezone.utc).isoformat() + + try: + session_duration_sec = max( + 0, int((window.end.micros - window.start.micros) / 1000000)) + except (OverflowError, ValueError): + session_duration_sec = 0 + + try: + start_sec = int(window.start.micros / 1000000) + except (OverflowError, ValueError): + start_sec = int(datetime.now(timezone.utc).timestamp()) + session_id = f"sess_{household_key}_{start_sec}" + processed_ts = datetime.now(timezone.utc).isoformat() + + transactions: List[Tuple[str, TransactionItem, Optional[str]]] = [] + coupons_by_tx: Dict[str, + List[CouponRedemption]] = collections.defaultdict(list) + + for ev in events: + if ev.transaction is not None: + transactions.append( + (ev.transaction_id, ev.transaction, ev.event_timestamp)) + elif ev.coupon is not None: + coupons_by_tx[ev.transaction_id].append(ev.coupon) + + distinct_tx_ids = set() + distinct_products = set() + campaigns = set() + stores = set() + total_spend = 0.0 + total_items = 0 + total_discount = 0.0 + coupons_redeemed_count = 0 + + for coupon_list in coupons_by_tx.values(): + coupons_redeemed_count += len(coupon_list) + for c in coupon_list: + if c.campaign: + campaigns.add(c.campaign) + + for tx_id, tx, event_ts in transactions: + distinct_tx_ids.add(tx_id) + if tx.product_id: + distinct_products.add(tx.product_id) + if tx.store_id: + stores.add(tx.store_id) + + total_spend += tx.sales_value + total_items += tx.quantity + total_discount += (tx.retail_disc + tx.coupon_disc) + + matching_coupons = coupons_by_tx.get(tx_id, []) + if not matching_coupons: + coupon_iter: List[Optional[CouponRedemption]] = [None] + else: + coupon_iter = matching_coupons + self.matched_coupons_counter.inc(len(matching_coupons)) + + for coup in coupon_iter: + coupon_upc = coup.coupon_upc if coup else None + campaign = coup.campaign if coup else None + + unified_record = UnifiedTransactionRecord( + session_id=session_id, + transaction_id=tx_id, + household_key=household_key, + product_id=tx.product_id, + quantity=tx.quantity, + sales_value=tx.sales_value, + store_id=tx.store_id, + retail_disc=tx.retail_disc, + coupon_discount=tx.coupon_disc, + coupon_match_disc=tx.coupon_match_disc, + coupon_upc=coupon_upc, + campaign=campaign, + day=tx.day, + trans_time=tx.trans_time, + week_no=tx.week_no, + event_timestamp=event_ts or processed_ts, + processed_timestamp=processed_ts, + ) + yield unified_record + self.unified_items_counter.inc() + + session_profile = CustomerSessionProfile( + session_id=session_id, + household_key=household_key, + session_start=session_start_iso, + session_end=session_end_iso, + session_duration_sec=session_duration_sec, + total_transactions=len(distinct_tx_ids), + total_items_purchased=total_items, + total_spend=round(total_spend, 2), + total_discount=round(total_discount, 2), + coupons_redeemed_count=coupons_redeemed_count, + distinct_products_count=len(distinct_products), + campaigns=sorted(list(campaigns)), + stores_visited=sorted(list(stores)), + processed_timestamp=processed_ts, + ) + yield beam.pvalue.TaggedOutput(TAG_SESSIONS, session_profile) + self.sessions_counter.inc() diff --git a/pipelines/cdp/cdp_pipeline/sinks.py b/pipelines/cdp/cdp_pipeline/sinks.py new file mode 100644 index 00000000..36696c6b --- /dev/null +++ b/pipelines/cdp/cdp_pipeline/sinks.py @@ -0,0 +1,160 @@ +# 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. +"""BigQuery sinks using the Storage Write API with auto-sharding.""" + +from datetime import datetime +import time +from typing import Any, Dict, Optional + +import apache_beam as beam +from apache_beam import PCollection +from apache_beam.io.gcp.bigquery import BigQueryDisposition, WriteToBigQuery +from apache_beam.options.pipeline_options import GoogleCloudOptions +from apache_beam.utils.timestamp import Timestamp + +from cdp_pipeline.options import MyPipelineOptions +from cdp_pipeline.schemas import load_output_schema + + +def _to_beam_timestamp(val: Any) -> Optional[Timestamp]: + """Converts string, numeric, or datetime timestamp into Beam Timestamp. + + Required for BigQuery Storage Write API compatibility. + """ + if val is None: + return None + if isinstance(val, Timestamp): + return val + if isinstance(val, (int, float)): + return Timestamp.of(float(val)) + if isinstance(val, datetime): + return Timestamp.of(val.timestamp()) + if isinstance(val, str): + try: + dt = datetime.fromisoformat(val.replace("Z", "+00:00")) + return Timestamp.of(dt.timestamp()) + except (ValueError, TypeError): + return Timestamp.of(time.time()) + return Timestamp.of(time.time()) + + +def _format_unified_dict(record: Any, use_storage_api: bool) -> Dict[str, Any]: + row = record.to_dict() if hasattr(record, "to_dict") else dict(record) + if use_storage_api: + if "event_timestamp" in row and row["event_timestamp"]: + row["event_timestamp"] = _to_beam_timestamp(row["event_timestamp"]) + if "processed_timestamp" in row and row["processed_timestamp"]: + row["processed_timestamp"] = _to_beam_timestamp( + row["processed_timestamp"]) + return row + + +def _format_session_dict(record: Any, use_storage_api: bool) -> Dict[str, Any]: + row = record.to_dict() if hasattr(record, "to_dict") else dict(record) + if use_storage_api: + if "session_start" in row and row["session_start"]: + row["session_start"] = _to_beam_timestamp(row["session_start"]) + if "session_end" in row and row["session_end"]: + row["session_end"] = _to_beam_timestamp(row["session_end"]) + if "processed_timestamp" in row and row["processed_timestamp"]: + row["processed_timestamp"] = _to_beam_timestamp( + row["processed_timestamp"]) + return row + + +def _format_deadletter_dict(record: Any, + use_storage_api: bool) -> Dict[str, Any]: + row = record.to_dict() if hasattr(record, "to_dict") else dict(record) + if use_storage_api: + if "timestamp" in row and row["timestamp"]: + row["timestamp"] = _to_beam_timestamp(row["timestamp"]) + return row + + +def apply_bigquery_sinks( + unified_records: PCollection, + customer_sessions: PCollection, + all_deadletters: PCollection, + pipeline_options: MyPipelineOptions, +) -> None: + """Configures BigQuery Storage Write API sinks for unified, session, and DLQ records.""" + project_id = pipeline_options.view_as(GoogleCloudOptions).project + dataset = getattr(pipeline_options, "output_dataset", "cdp_dataset") + unified_table = getattr(pipeline_options, "output_table", + "unified_customer_data") + sessions_table = getattr(pipeline_options, "output_sessions_table", + "customer_sessions") + deadletter_table = getattr(pipeline_options, "deadletter_table", None) + use_storage_api = getattr(pipeline_options, "use_storage_write_api", True) + + write_method = ( + WriteToBigQuery.Method.STORAGE_WRITE_API + if use_storage_api else WriteToBigQuery.Method.STREAMING_INSERTS) + + if not (project_id and dataset): + return + + unified_schema = load_output_schema( + getattr(pipeline_options, "output_schema_path", None), + "unified_table.json", + ) + sessions_schema = load_output_schema( + getattr(pipeline_options, "output_sessions_schema_path", None), + "customer_sessions.json", + ) + + unified_table_spec = f"{project_id}:{dataset}.{unified_table}" + (unified_records + | "Format Unified Rows" >> beam.Map(_format_unified_dict, use_storage_api) + | "Write Unified to BigQuery" >> WriteToBigQuery( + table=unified_table_spec, + schema=unified_schema, + create_disposition=BigQueryDisposition.CREATE_IF_NEEDED, + write_disposition=BigQueryDisposition.WRITE_APPEND, + method=write_method, + with_auto_sharding=True if use_storage_api else False, + triggering_frequency=5 if use_storage_api else None, + )) + + sessions_table_spec = f"{project_id}:{dataset}.{sessions_table}" + (customer_sessions + | "Format Sessions Rows" >> beam.Map(_format_session_dict, use_storage_api) + | "Write Sessions to BigQuery" >> WriteToBigQuery( + table=sessions_table_spec, + schema=sessions_schema, + create_disposition=BigQueryDisposition.CREATE_IF_NEEDED, + write_disposition=BigQueryDisposition.WRITE_APPEND, + method=write_method, + with_auto_sharding=True if use_storage_api else False, + triggering_frequency=5 if use_storage_api else None, + )) + + if deadletter_table: + deadletter_schema = load_output_schema( + getattr(pipeline_options, "deadletter_schema_path", None), + "deadletter_table.json", + ) + dlq_table_spec = f"{project_id}:{dataset}.{deadletter_table}" + (all_deadletters + | "Format Deadletter Rows" >> beam.Map(_format_deadletter_dict, + use_storage_api) + | "Write Deadletter to BigQuery" >> WriteToBigQuery( + table=dlq_table_spec, + schema=deadletter_schema, + create_disposition=BigQueryDisposition.CREATE_IF_NEEDED, + write_disposition=BigQueryDisposition.WRITE_APPEND, + method=write_method, + with_auto_sharding=True if use_storage_api else False, + triggering_frequency=5 if use_storage_api else None, + )) diff --git a/pipelines/cdp/cloudbuild.yaml b/pipelines/cdp/cloudbuild.yaml index d4af5663..29e35a40 100644 --- a/pipelines/cdp/cloudbuild.yaml +++ b/pipelines/cdp/cloudbuild.yaml @@ -13,13 +13,13 @@ # limitations under the License. steps: - - name: 'gcr.io/cloud-builders/docker' - script: | - docker build -t ${_TAG} . + - name: 'gcr.io/kaniko-project/executor:latest' + args: + - --destination=${_TAG} + - --cache=true substitutions: - _TAG: unset + _TAG: unset options: - substitutionOption: 'ALLOW_LOOSE' - automapSubstitutions: true -images: - - ${_TAG} \ No newline at end of file + substitutionOption: 'ALLOW_LOOSE' + automapSubstitutions: true + machineType: E2_HIGHCPU_8 \ No newline at end of file diff --git a/pipelines/cdp/main.py b/pipelines/cdp/main.py index 77d3892a..4e58901e 100644 --- a/pipelines/cdp/main.py +++ b/pipelines/cdp/main.py @@ -20,7 +20,7 @@ from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions from cdp_pipeline.options import MyPipelineOptions -from cdp_pipeline.customer_data_platform import create_and_run_pipeline +from cdp_pipeline.pipeline import create_and_run_pipeline def main(options: MyPipelineOptions): @@ -32,7 +32,6 @@ def main(options: MyPipelineOptions): dataflow_options: GoogleCloudOptions = pipeline_options.view_as( GoogleCloudOptions) now_epoch_ms = int(time.time() * 1000) - dataflow_options.job_name = f"customer-data-platform-{now_epoch_ms}" - custom_options: MyPipelineOptions = pipeline_options.view_as( - MyPipelineOptions) - main(custom_options) + if not dataflow_options.job_name: + dataflow_options.job_name = f"customer-data-platform-{now_epoch_ms}" + main(pipeline_options.view_as(MyPipelineOptions)) diff --git a/pipelines/cdp/pytest.ini b/pipelines/cdp/pytest.ini new file mode 100644 index 00000000..64abdf22 --- /dev/null +++ b/pipelines/cdp/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +addopts = -p no:locust diff --git a/pipelines/cdp/requirements.txt b/pipelines/cdp/requirements.txt index 744d2094..f7b195f5 100644 --- a/pipelines/cdp/requirements.txt +++ b/pipelines/cdp/requirements.txt @@ -13,7 +13,7 @@ # limitations under the License. apache-beam[gcp]==2.76.0 # Example, use your actual versions -## Below dependencies are required if you have to run script /cdp_pipeline/generate_transaction_data.py +## Below dependencies are required if you run scripts/03_publish_events.py or simulator pandas fsspec gcsfs diff --git a/pipelines/cdp/schema/customer_sessions.json b/pipelines/cdp/schema/customer_sessions.json new file mode 100644 index 00000000..ded64701 --- /dev/null +++ b/pipelines/cdp/schema/customer_sessions.json @@ -0,0 +1,74 @@ +{ + "fields": [ + { + "name": "session_id", + "type": "STRING", + "mode": "REQUIRED" + }, + { + "name": "household_key", + "type": "STRING", + "mode": "REQUIRED" + }, + { + "name": "session_start", + "type": "TIMESTAMP", + "mode": "REQUIRED" + }, + { + "name": "session_end", + "type": "TIMESTAMP", + "mode": "REQUIRED" + }, + { + "name": "session_duration_sec", + "type": "INTEGER", + "mode": "REQUIRED" + }, + { + "name": "total_transactions", + "type": "INTEGER", + "mode": "REQUIRED" + }, + { + "name": "total_items_purchased", + "type": "INTEGER", + "mode": "REQUIRED" + }, + { + "name": "total_spend", + "type": "FLOAT", + "mode": "REQUIRED" + }, + { + "name": "total_discount", + "type": "FLOAT", + "mode": "REQUIRED" + }, + { + "name": "coupons_redeemed_count", + "type": "INTEGER", + "mode": "REQUIRED" + }, + { + "name": "distinct_products_count", + "type": "INTEGER", + "mode": "REQUIRED" + }, + { + "name": "campaigns", + "type": "STRING", + "mode": "REPEATED" + }, + { + "name": "stores_visited", + "type": "STRING", + "mode": "REPEATED" + }, + { + "name": "processed_timestamp", + "type": "TIMESTAMP", + "mode": "REQUIRED" + } + ] +} diff --git a/pipelines/cdp/schema/deadletter_table.json b/pipelines/cdp/schema/deadletter_table.json new file mode 100644 index 00000000..d87cd242 --- /dev/null +++ b/pipelines/cdp/schema/deadletter_table.json @@ -0,0 +1,24 @@ +{ + "fields": [ + { + "name": "source", + "type": "STRING", + "mode": "REQUIRED" + }, + { + "name": "raw_payload", + "type": "STRING", + "mode": "NULLABLE" + }, + { + "name": "error_message", + "type": "STRING", + "mode": "REQUIRED" + }, + { + "name": "timestamp", + "type": "TIMESTAMP", + "mode": "REQUIRED" + } + ] +} diff --git a/pipelines/cdp/schema/unified_table.json b/pipelines/cdp/schema/unified_table.json index 545fa996..20683db8 100644 --- a/pipelines/cdp/schema/unified_table.json +++ b/pipelines/cdp/schema/unified_table.json @@ -1,29 +1,89 @@ { - "fields": [ - { - "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" - } - ] + "fields": [ + { + "name": "session_id", + "type": "STRING", + "mode": "REQUIRED" + }, + { + "name": "transaction_id", + "type": "STRING", + "mode": "REQUIRED" + }, + { + "name": "household_key", + "type": "STRING", + "mode": "REQUIRED" + }, + { + "name": "product_id", + "type": "STRING", + "mode": "NULLABLE" + }, + { + "name": "quantity", + "type": "INTEGER", + "mode": "NULLABLE" + }, + { + "name": "sales_value", + "type": "FLOAT", + "mode": "NULLABLE" + }, + { + "name": "store_id", + "type": "STRING", + "mode": "NULLABLE" + }, + { + "name": "retail_disc", + "type": "FLOAT", + "mode": "NULLABLE" + }, + { + "name": "coupon_discount", + "type": "FLOAT", + "mode": "NULLABLE" + }, + { + "name": "coupon_match_disc", + "type": "FLOAT", + "mode": "NULLABLE" + }, + { + "name": "coupon_upc", + "type": "STRING", + "mode": "NULLABLE" + }, + { + "name": "campaign", + "type": "STRING", + "mode": "NULLABLE" + }, + { + "name": "day", + "type": "INTEGER", + "mode": "NULLABLE" + }, + { + "name": "trans_time", + "type": "STRING", + "mode": "NULLABLE" + }, + { + "name": "week_no", + "type": "INTEGER", + "mode": "NULLABLE" + }, + { + "name": "event_timestamp", + "type": "TIMESTAMP", + "mode": "NULLABLE" + }, + { + "name": "processed_timestamp", + "type": "TIMESTAMP", + "mode": "REQUIRED" + } + ] } \ No newline at end of file diff --git a/pipelines/cdp/scripts/02_run_dataflow.sh b/pipelines/cdp/scripts/02_run_dataflow.sh index 165a47b4..c42c8dd3 100755 --- a/pipelines/cdp/scripts/02_run_dataflow.sh +++ b/pipelines/cdp/scripts/02_run_dataflow.sh @@ -16,12 +16,20 @@ 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 +python_version=$(python3 -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")' 2>/dev/null || true) +if [[ "$python_version" != "3.14" ]]; then + echo "Error: Python 3.14 is required to launch Dataflow, but active python is '$python_version'." >&2 + echo "Please activate your Python 3.14 virtual environment (e.g. 'source ~/.virtualenvs/cdp314/bin/activate')." >&2 + exit 1 +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}" @@ -34,6 +42,26 @@ elif [ -n "$NETWORK" ]; then SUBNET_OPT="--subnetwork=$NETWORK" fi +INPUT_ARGS=() +if [ -n "$TRANSACTIONS_SUBSCRIPTION" ]; then + INPUT_ARGS+=(--transactions_subscription="$TRANSACTIONS_SUBSCRIPTION") +elif [ -n "$TRANSACTIONS_TOPIC" ]; then + INPUT_ARGS+=(--transactions_topic="$TRANSACTIONS_TOPIC") +fi + +if [ -n "$COUPON_REDEMPTION_SUBSCRIPTION" ]; then + INPUT_ARGS+=(--coupons_redemption_subscription="$COUPON_REDEMPTION_SUBSCRIPTION") +elif [ -n "$COUPON_REDEMPTION_TOPIC" ]; then + INPUT_ARGS+=(--coupons_redemption_topic="$COUPON_REDEMPTION_TOPIC") +fi + +DLQ_ARGS=() +if [ -n "$BQ_DEADLETTER_TABLE" ]; then + DLQ_ARGS+=(--deadletter_table="$BQ_DEADLETTER_TABLE") +fi + +cd "$PIPELINE_DIR" + echo "Submitting Customer Data Platform Dataflow pipeline..." python3 -m main \ --streaming \ @@ -42,16 +70,21 @@ python3 -m main \ --temp_location="${TEMP_LOCATION:-gs://$PROJECT/tmp}" \ --region="$REGION" \ --save_main_session \ + --setup_file=./setup.py \ --service_account_email="$SERVICE_ACCOUNT" \ $SUBNET_OPT \ --no_use_public_ips \ --sdk_container_image="$CONTAINER_URI" \ + --sdk_location=container \ --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" \ + "${INPUT_ARGS[@]}" \ --output_dataset="$BQ_DATASET" \ --output_table="$BQ_UNIFIED_TABLE" \ - --project_id="$PROJECT" \ + --output_sessions_table="${BQ_SESSIONS_TABLE:-customer_sessions}" \ + "${DLQ_ARGS[@]}" \ + --session_gap_seconds="${SESSION_GAP_SECONDS:-900}" \ + --allowed_lateness_seconds="${ALLOWED_LATENESS_SECONDS:-60}" \ + --use_storage_write_api \ --enable_streaming_engine diff --git a/pipelines/cdp/scripts/02_run_local.sh b/pipelines/cdp/scripts/02_run_local.sh new file mode 100755 index 00000000..b3e9e2f8 --- /dev/null +++ b/pipelines/cdp/scripts/02_run_local.sh @@ -0,0 +1,40 @@ +#!/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. + +# Runs the Customer Data Platform pipeline locally with DirectRunner for testing and development + +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 + +cd "$PIPELINE_DIR" + +python3 -m main \ + --runner=DirectRunner \ + --project="${PROJECT:-local-test-project}" \ + --temp_location=/tmp/dataflow-temp \ + --transactions_topic="${TRANSACTIONS_TOPIC:-projects/local-test-project/topics/cdp-transactions}" \ + --coupons_redemption_topic="${COUPON_REDEMPTION_TOPIC:-projects/local-test-project/topics/cdp-coupon-redemption}" \ + --output_dataset="${BQ_DATASET:-cdp_dataset}" \ + --output_table="${BQ_UNIFIED_TABLE:-unified_customer_data}" \ + --output_sessions_table="${BQ_SESSIONS_TABLE:-customer_sessions}" \ + --session_gap_seconds=10 \ + --allowed_lateness_seconds=5 diff --git a/pipelines/cdp/scripts/03_publish_events.py b/pipelines/cdp/scripts/03_publish_events.py new file mode 100755 index 00000000..6e20b6c7 --- /dev/null +++ b/pipelines/cdp/scripts/03_publish_events.py @@ -0,0 +1,31 @@ +#!/usr/bin/env python3 +# 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. +"""CLI launcher script to publish synthetic Customer Data Platform streaming events to Pub/Sub.""" + +# pylint: disable=invalid-name,wrong-import-position + +import os +import sys + +# Ensure pipelines/cdp root is on Python path +CURRENT_DIR = os.path.dirname(os.path.abspath(__file__)) +CDP_ROOT = os.path.dirname(CURRENT_DIR) +if CDP_ROOT not in sys.path: + sys.path.insert(0, CDP_ROOT) + +from simulator.publisher import main + +if __name__ == "__main__": + main() diff --git a/pipelines/cdp/setup.py b/pipelines/cdp/setup.py index e1860b94..7501462e 100644 --- a/pipelines/cdp/setup.py +++ b/pipelines/cdp/setup.py @@ -18,12 +18,17 @@ from setuptools import setup, find_packages with open("requirements.txt", encoding="utf-8") as f: - requirements = f.readlines() + requirements = [ + line.strip() + for line in f + if line.strip() and not line.strip().startswith("#") + ] setup( - name="Dataflow Solution for Customer Data Platform", + name="cdp_pipeline", version="0.1", description="Customer Data Platform example for the Dataflow Solution Guides", packages=find_packages(), + include_package_data=True, install_requires=requirements, ) diff --git a/pipelines/cdp/simulator/__init__.py b/pipelines/cdp/simulator/__init__.py new file mode 100644 index 00000000..b4354dea --- /dev/null +++ b/pipelines/cdp/simulator/__init__.py @@ -0,0 +1,14 @@ +# 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. +"""Customer Data Platform test data simulator package.""" diff --git a/pipelines/cdp/simulator/generator.py b/pipelines/cdp/simulator/generator.py new file mode 100644 index 00000000..ff8f3266 --- /dev/null +++ b/pipelines/cdp/simulator/generator.py @@ -0,0 +1,125 @@ +# 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. +"""Synthetic customer session data generator for testing and simulations.""" + +from datetime import datetime, timedelta, timezone +import random +from typing import Any, Dict, List, Tuple + +from cdp_pipeline.models import ( + CouponRedemption, + CustomerInteractionEvent, + EventType, + TransactionItem, +) + + +def generate_synthetic_session_events( + household_key: str, + base_tx_id: int, + session_offset_sec: int = 0, +) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: + """Generates realistic transaction items and coupons for a customer shopping session.""" + tx_id = str(base_tx_id) + event_time = datetime.now( + timezone.utc) - timedelta(seconds=session_offset_sec) + now_iso = event_time.isoformat() + store_id = str(random.choice([101, 204, 305, 436])) + + # Products in session basket + products = [ + { + "id": "941769", + "price": 3.99, + "qty": 1, + "disc": 0.50 + }, + { + "id": "910635", + "price": 2.99, + "qty": 2, + "disc": 0.00 + }, + { + "id": "1082185", + "price": 1.49, + "qty": 1, + "disc": 0.25 + }, + ] + selected = random.sample(products, k=random.randint(1, len(products))) + + transactions = [] + for p in selected: + tx_item = TransactionItem( + product_id=p["id"], + quantity=p["qty"], + sales_value=round(p["price"] * p["qty"], 2), + store_id=store_id, + retail_disc=0.0, + coupon_disc=p["disc"], + coupon_match_disc=0.0, + day=421, + week_no=8, + trans_time="1456", + ) + tx_event = CustomerInteractionEvent( + event_type=EventType.TRANSACTION.value, + household_key=household_key, + transaction_id=tx_id, + event_timestamp=now_iso, + transaction=tx_item, + coupon=None, + ) + transactions.append({ + "household_key": tx_event.household_key, + "transaction_id": tx_event.transaction_id, + "product_id": tx_item.product_id, + "quantity": tx_item.quantity, + "sales_value": tx_item.sales_value, + "store_id": tx_item.store_id, + "retail_disc": tx_item.retail_disc, + "coupon_disc": tx_item.coupon_disc, + "coupon_match_disc": tx_item.coupon_match_disc, + "day": tx_item.day, + "week_no": tx_item.week_no, + "trans_time": tx_item.trans_time, + "event_timestamp": tx_event.event_timestamp, + }) + + coupons = [] + if random.random() < 0.7: # 70% chance of coupon redemption + coupon_item = CouponRedemption( + coupon_upc=str(random.choice([10000085364, 51700010076, 10000089277])), + campaign=str(random.choice([2200, 18, 500])), + day=421, + ) + cp_event = CustomerInteractionEvent( + event_type=EventType.COUPON.value, + household_key=household_key, + transaction_id=tx_id, + event_timestamp=now_iso, + transaction=None, + coupon=coupon_item, + ) + coupons.append({ + "household_key": cp_event.household_key, + "transaction_id": cp_event.transaction_id, + "coupon_upc": coupon_item.coupon_upc, + "campaign": coupon_item.campaign, + "day": coupon_item.day, + "event_timestamp": cp_event.event_timestamp, + }) + + return transactions, coupons diff --git a/pipelines/cdp/simulator/publisher.py b/pipelines/cdp/simulator/publisher.py new file mode 100644 index 00000000..94a92771 --- /dev/null +++ b/pipelines/cdp/simulator/publisher.py @@ -0,0 +1,158 @@ +# 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. +"""Asynchronous Google Cloud Pub/Sub streaming publisher for Customer Data Platform.""" + +import argparse +import asyncio +import json +import logging +import os +import random +from typing import Optional + +from google.cloud import pubsub_v1 + +from simulator.generator import generate_synthetic_session_events + + +def get_topic_path(publisher: pubsub_v1.PublisherClient, project: str, + topic: str) -> str: + """Returns a fully-qualified Pub/Sub topic path.""" + if topic.startswith("projects/"): + return topic + return publisher.topic_path(project, topic) + + +async def publish_events_to_pubsub( + project_id: Optional[str] = None, + transactions_topic: Optional[str] = None, + coupons_topic: Optional[str] = None, + continuous: bool = False, + interval: float = 1.0, + count: int = 10, + inject_errors: bool = False, +): + """Publishes sessionized transactions and coupon redemptions to Pub/Sub topics.""" + project_id = project_id or os.environ.get("PROJECT", "") + tx_topic_name = transactions_topic or os.environ.get("TRANSACTIONS_TOPIC", + "cdp-transactions") + cp_topic_name = coupons_topic or os.environ.get("COUPON_REDEMPTION_TOPIC", + "cdp-coupon-redemption") + + publisher = pubsub_v1.PublisherClient() + tx_path = get_topic_path(publisher, project_id, tx_topic_name) + cp_path = get_topic_path(publisher, project_id, cp_topic_name) + + logging.info("Publishing transactions to: %s", tx_path) + logging.info("Publishing coupons to: %s", cp_path) + + households = ["1", "13", "42", "99", "125"] + tx_counter = 27601281000 + published_count = 0 + + while True: + hh = random.choice(households) + tx_counter += 1 + tx_items, cp_items = generate_synthetic_session_events(hh, tx_counter) + + # Publish transactions + for tx in tx_items: + payload = json.dumps(tx).encode("utf-8") + future = publisher.publish(tx_path, payload) + logging.info("Published tx [hh=%s, tx=%s]: msg_id=%s", hh, + tx["transaction_id"], future.result()) + + # Publish coupons + for cp in cp_items: + payload = json.dumps(cp).encode("utf-8") + future = publisher.publish(cp_path, payload) + logging.info("Published coupon [hh=%s, tx=%s]: msg_id=%s", hh, + cp["transaction_id"], future.result()) + + # Error injection test (DLQ verification) + if inject_errors and random.random() < 0.2: + corrupt_payload = b"NOT_VALID_JSON_{broken: true" + future = publisher.publish(tx_path, corrupt_payload) + logging.info("Injected malformed transaction DLQ test payload: msg_id=%s", + future.result()) + + published_count += len(tx_items) + len(cp_items) + if not continuous and published_count >= count: + break + + await asyncio.sleep(interval) + + +def main(): + """Parses command line arguments and runs the Pub/Sub publishing loop.""" + logging.basicConfig( + level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") + parser = argparse.ArgumentParser( + description="Publish Customer Data Platform sessions and events to Pub/Sub." + ) + parser.add_argument( + "--project", + "--project_id", + dest="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 path", + ) + parser.add_argument( + "--coupons_topic", + default=os.environ.get("COUPON_REDEMPTION_TOPIC"), + help="Coupons Pub/Sub topic name or path", + ) + parser.add_argument( + "--continuous", + action="store_true", + help="Continuously stream synthetic sessions until cancelled", + ) + parser.add_argument( + "--interval", + type=float, + default=1.0, + help="Sleep interval in seconds between published customer sessions", + ) + parser.add_argument( + "--count", + type=int, + default=20, + help="Total events to publish when not running continuously", + ) + parser.add_argument( + "--inject_errors", + action="store_true", + help="Inject malformed payloads to verify Dead-Letter Queue (DLQ) processing", + ) + args = parser.parse_args() + + asyncio.run( + publish_events_to_pubsub( + project_id=args.project_id, + transactions_topic=args.transactions_topic, + coupons_topic=args.coupons_topic, + continuous=args.continuous, + interval=args.interval, + count=args.count, + inject_errors=args.inject_errors, + )) + + +if __name__ == "__main__": + main() diff --git a/pipelines/cdp/tests/test_customer_data_platform.py b/pipelines/cdp/tests/test_customer_data_platform.py index 80a63684..2cbb6ccf 100644 --- a/pipelines/cdp/tests/test_customer_data_platform.py +++ b/pipelines/cdp/tests/test_customer_data_platform.py @@ -11,77 +11,53 @@ # 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. -""" -Unit tests for Customer Data Platform pipeline transformations. -""" +"""Unit tests for Customer Data Platform pipeline transformations and sessionization.""" +import json import unittest -from apache_beam.testing.test_pipeline import TestPipeline -from apache_beam.testing.util import assert_that, equal_to + import apache_beam as beam +from apache_beam.options.pipeline_options import GoogleCloudOptions +from apache_beam.testing.test_pipeline import TestPipeline +from apache_beam.testing.util import assert_that +from apache_beam.transforms.window import IntervalWindow +from apache_beam.typehints.schemas import named_tuple_to_schema +from apache_beam.utils.timestamp import Timestamp -from cdp_pipeline.customer_data_platform import ( - left_join, - load_output_schema, - _unify_data, +from cdp_pipeline.models import ( + CouponRedemption, + CustomerInteractionEvent, + CustomerSessionProfile, + DeadLetterRecord, + EventType, + TransactionItem, + UnifiedTransactionRecord, +) +from cdp_pipeline.options import MyPipelineOptions +from cdp_pipeline.parsing import ( + ParseRecordDoFn, + TAG_DEADLETTER, +) +from cdp_pipeline.pipeline import build_pipeline +from cdp_pipeline.schemas import load_output_schema +from cdp_pipeline.sessionization import ( + ProcessCustomerSessionDoFn, + TAG_SESSIONS, ) +# Prevent pytest from treating Apache Beam's TestPipeline as a test case +TestPipeline.__test__ = False -class CustomerDataPlatformTest(unittest.TestCase): - def test_left_join_with_matching_coupons(self): - key = ("27601281299", "1") - transactions = [{ - "transaction_id": "27601281299", - "household_key": "1", - "product_id": "941769", - "coupon_disc": "0.50", - }] - coupons = [{ - "transaction_id": "27601281299", - "household_key": "1", - "coupon_upc": "10000085364", - "campaign": "2200", - }] - - results = list(left_join((key, (transactions, coupons)))) - self.assertEqual(len(results), 1) - self.assertEqual( - results[0], - { - "transaction_id": "27601281299", - "household_key": "1", - "coupon_upc": "10000085364", - "product_id": "941769", - "coupon_discount": "0.50", - }, - ) +class CustomerDataPlatformTest(unittest.TestCase): - def test_left_join_without_matching_coupons(self): - key = ("27601281299", "1") - transactions = [{ - "transaction_id": "27601281299", - "household_key": "1", - "product_id": "941769", - "coupon_disc": "0", - }] - coupons = [] - - results = list(left_join((key, (transactions, coupons)))) - self.assertEqual(len(results), 1) - self.assertEqual( - results[0], - { - "transaction_id": "27601281299", - "household_key": "1", - "coupon_upc": None, - "product_id": "941769", - "coupon_discount": "0", - }, - ) + def test_pipeline_options_project(self): + options = MyPipelineOptions(["--project=my-test-project"]) + gcp_options = options.view_as(GoogleCloudOptions) + self.assertEqual(gcp_options.project, "my-test-project") def test_load_output_schema_default(self): - schema = load_output_schema(None) + schema = load_output_schema() self.assertIn("fields", schema) field_names = [field["name"] for field in schema["fields"]] self.assertIn("transaction_id", field_names) @@ -89,52 +65,377 @@ def test_load_output_schema_default(self): self.assertIn("coupon_upc", field_names) self.assertIn("product_id", field_names) self.assertIn("coupon_discount", field_names) + self.assertIn("session_id", field_names) + self.assertIn("campaign", field_names) + + def test_load_schemas_helpers(self): + sessions_schema = load_output_schema(None, "customer_sessions.json") + self.assertIn("fields", sessions_schema) + session_fields = [f["name"] for f in sessions_schema["fields"]] + self.assertIn("session_id", session_fields) + self.assertIn("total_spend", session_fields) + self.assertIn("total_transactions", session_fields) + + dlq_schema = load_output_schema(None, "deadletter_table.json") + self.assertIn("fields", dlq_schema) + dlq_fields = [f["name"] for f in dlq_schema["fields"]] + self.assertIn("error_message", dlq_fields) + self.assertIn("raw_payload", dlq_fields) + + with self.assertRaises(FileNotFoundError): + load_output_schema(None, "non_existent_schema.json") + + def test_parse_record_valid_transaction(self): + fn = ParseRecordDoFn(EventType.TRANSACTION) + fn.setup() + raw = json.dumps({ + "household_key": "100", + "transaction_id": "tx-1", + "product_id": "prod-1", + "sales_value": 15.50 + }).encode("utf-8") + + results = list(fn.process(raw)) + self.assertEqual(len(results), 1) + hh_key, event = results[0] + self.assertEqual(hh_key, "100") + self.assertIsInstance(event, CustomerInteractionEvent) + self.assertEqual(event.transaction_id, "tx-1") + self.assertEqual(event.event_type, EventType.TRANSACTION.value) + self.assertIsNotNone(event.transaction) + self.assertEqual(event.transaction.product_id, "prod-1") + self.assertEqual(event.transaction.sales_value, 15.50) + self.assertIsNone(event.coupon) + + def test_parse_record_valid_coupon(self): + fn = ParseRecordDoFn(EventType.COUPON) + fn.setup() + raw = json.dumps({ + "household_key": "100", + "transaction_id": "tx-1", + "coupon_upc": "cp-1", + "campaign": "camp-99" + }) + + results = list(fn.process(raw)) + self.assertEqual(len(results), 1) + hh_key, event = results[0] + self.assertEqual(hh_key, "100") + self.assertIsInstance(event, CustomerInteractionEvent) + self.assertEqual(event.transaction_id, "tx-1") + self.assertEqual(event.event_type, EventType.COUPON.value) + self.assertIsNotNone(event.coupon) + self.assertEqual(event.coupon.coupon_upc, "cp-1") + self.assertEqual(event.coupon.campaign, "camp-99") + self.assertIsNone(event.transaction) + + def test_parse_record_malformed_json_dlq(self): + fn = ParseRecordDoFn("transaction") + fn.setup() + bad_bytes = b"BROKEN_JSON_DATA{{{" + + results = list(fn.process(bad_bytes)) + self.assertEqual(len(results), 1) + tagged_output = results[0] + self.assertIsInstance(tagged_output, beam.pvalue.TaggedOutput) + self.assertEqual(tagged_output.tag, TAG_DEADLETTER) + self.assertIn("Malformed payload", tagged_output.value["error_message"]) + + def test_parse_record_missing_keys_dlq(self): + fn = ParseRecordDoFn("transaction") + fn.setup() + # Missing household_key + missing_key = json.dumps({"transaction_id": "tx-1"}).encode("utf-8") + + results = list(fn.process(missing_key)) + self.assertEqual(len(results), 1) + tagged_output = results[0] + self.assertIsInstance(tagged_output, beam.pvalue.TaggedOutput) + self.assertEqual(tagged_output.tag, TAG_DEADLETTER) + self.assertIn("Missing required household_key", + tagged_output.value["error_message"]) + + def test_process_customer_session_aggregation(self): + fn = ProcessCustomerSessionDoFn() + fn.setup() + + household_key = "hh-42" + mock_window = IntervalWindow( # pylint: disable=too-many-function-args + Timestamp(1000), Timestamp(1300)) + + events = [ + CustomerInteractionEvent( + event_type=EventType.TRANSACTION.value, + household_key="hh-42", + transaction_id="tx-101", + transaction=TransactionItem( + product_id="prod-A", + quantity=2, + sales_value=20.0, + retail_disc=2.0, + coupon_disc=1.0, + store_id="store-1", + ), + ), + CustomerInteractionEvent( + event_type=EventType.TRANSACTION.value, + household_key="hh-42", + transaction_id="tx-102", + transaction=TransactionItem( + product_id="prod-B", + quantity=1, + sales_value=10.0, + retail_disc=0.0, + coupon_disc=0.0, + store_id="store-1", + ), + ), + CustomerInteractionEvent( + event_type=EventType.COUPON.value, + household_key="hh-42", + transaction_id="tx-101", + coupon=CouponRedemption( + coupon_upc="cp-999", + campaign="fall-sale", + ), + ), + ] + + outputs = list(fn.process((household_key, events), window=mock_window)) - def test_unify_data_transform(self): - transactions_input = [ - (("t1", "h1"), { - "transaction_id": "t1", - "household_key": "h1", - "product_id": "p1", - "coupon_disc": "1.0", - }), - (("t2", "h2"), { - "transaction_id": "t2", - "household_key": "h2", - "product_id": "p2", - "coupon_disc": "0.0", - }), + # Main outputs: unified transaction records + unified = [ + item for item in outputs + if not isinstance(item, beam.pvalue.TaggedOutput) ] - coupons_input = [ - (("t1", "h1"), { - "transaction_id": "t1", - "household_key": "h1", - "coupon_upc": "c1", - }), + # Tagged output: Customer 360 session profile + sessions = [ + item.value for item in outputs if + isinstance(item, beam.pvalue.TaggedOutput) and item.tag == TAG_SESSIONS ] - expected = [ - { - "transaction_id": "t1", - "household_key": "h1", - "coupon_upc": "c1", - "product_id": "p1", - "coupon_discount": "1.0", - }, - { - "transaction_id": "t2", - "household_key": "h2", - "coupon_upc": None, - "product_id": "p2", - "coupon_discount": "0.0", - }, + # Two transaction items were emitted + self.assertEqual(len(unified), 2) + self.assertIsInstance(unified[0], UnifiedTransactionRecord) + tx_101 = next(u for u in unified if u.transaction_id == "tx-101") + self.assertEqual(tx_101.coupon_upc, "cp-999") + self.assertEqual(tx_101.campaign, "fall-sale") + self.assertEqual(tx_101.sales_value, 20.0) + + tx_102 = next(u for u in unified if u.transaction_id == "tx-102") + self.assertIsNone(tx_102.coupon_upc) + + # Verify session summary + self.assertEqual(len(sessions), 1) + session = sessions[0] + self.assertIsInstance(session, CustomerSessionProfile) + self.assertEqual(session.household_key, "hh-42") + self.assertEqual(session.total_transactions, 2) + self.assertEqual(session.total_items_purchased, 3) + self.assertEqual(session.total_spend, 30.0) + self.assertEqual(session.total_discount, 3.0) + self.assertEqual(session.coupons_redeemed_count, 1) + self.assertEqual(session.distinct_products_count, 2) + self.assertIn("fall-sale", session.campaigns) + + def test_build_pipeline_end_to_end_in_memory(self): + options = MyPipelineOptions( + session_gap_seconds=10, + allowed_lateness_seconds=5, + output_dataset="test_dataset", + output_table="test_unified", + output_sessions_table="test_sessions", + ) + + in_memory_tx = [ + json.dumps({ + "household_key": "hh-1", + "transaction_id": "tx-1", + "product_id": "p100", + "quantity": 1, + "sales_value": 5.0, + "event_timestamp": "2026-09-08T10:00:00Z", + }).encode("utf-8"), + b"MALFORMED_JSON_PAYLOAD", + ] + in_memory_cp = [ + json.dumps({ + "household_key": "hh-1", + "transaction_id": "tx-1", + "coupon_upc": "c100", + "campaign": "summer", + "event_timestamp": "2026-09-08T10:00:02Z", + }).encode("utf-8") ] - with TestPipeline() as p: - tx_pcoll = p | "Create Transactions" >> beam.Create(transactions_input) - cp_pcoll = p | "Create Coupons" >> beam.Create(coupons_input) - unified = (tx_pcoll, cp_pcoll) | _unify_data() - assert_that(unified, equal_to(expected)) + with TestPipeline(options=options) as p: + unified, sessions, deadletters = build_pipeline( + pipeline=p, + pipeline_options=options, + in_memory_transactions=in_memory_tx, + in_memory_coupons=in_memory_cp, + ) + + assert_that(unified, _check_unified, label="CheckUnified") + assert_that(sessions, _check_sessions, label="CheckSessions") + assert_that(deadletters, _check_deadletters, label="CheckDeadletters") + + +def _check_unified(records): + assert len(records) == 1 + assert records[0].transaction_id == "tx-1" + assert records[0].coupon_upc == "c100" + assert records[0].campaign == "summer" + + +def _check_sessions(session_records): + assert len(session_records) == 1 + assert session_records[0].household_key == "hh-1" + assert session_records[0].total_spend == 5.0 + + +def _check_deadletters(dlq_records): + assert len(dlq_records) == 1 + assert dlq_records[0]["source"] == "transaction" + assert "Malformed payload" in dlq_records[0]["error_message"] + + +class ModelsTest(unittest.TestCase): + """Unit tests for Beam schema data models and parsing logic.""" + + def test_transaction_item_defaults_and_dict(self): + item = TransactionItem( + product_id="prod-1", + quantity=3, + sales_value=25.50, + store_id="store-10", + retail_disc=1.50, + coupon_disc=0.50, + ) + self.assertEqual(item.product_id, "prod-1") + self.assertEqual(item.quantity, 3) + self.assertEqual(item.sales_value, 25.50) + item_dict = item.to_dict() + self.assertEqual(item_dict["product_id"], "prod-1") + self.assertEqual(item_dict["quantity"], 3) + self.assertEqual(item_dict["sales_value"], 25.50) + self.assertEqual(item_dict["store_id"], "store-10") + self.assertIsNone(item_dict["day"]) + + def test_coupon_redemption_defaults_and_dict(self): + coupon = CouponRedemption( + coupon_upc="cp-12345", + campaign="spring-sale", + day=15, + ) + self.assertEqual(coupon.coupon_upc, "cp-12345") + self.assertEqual(coupon.campaign, "spring-sale") + self.assertEqual(coupon.day, 15) + c_dict = coupon.to_dict() + self.assertEqual(c_dict["coupon_upc"], "cp-12345") + self.assertEqual(c_dict["campaign"], "spring-sale") + self.assertEqual(c_dict["day"], 15) + + def test_dead_letter_record_dict(self): + dlq = DeadLetterRecord( + source="transaction", + raw_payload='{"bad": "data"}', + error_message="Missing required field", + timestamp="2026-09-07T12:00:00Z", + ) + self.assertEqual(dlq.source, "transaction") + d_dict = dlq.to_dict() + self.assertEqual(d_dict["source"], "transaction") + self.assertEqual(d_dict["error_message"], "Missing required field") + + def test_customer_interaction_event_from_raw_payload_transaction(self): + payload = json.dumps({ + "household_key": "hh-99", + "transaction_id": "tx-888", + "product_id": "prod-abc", + "quantity": "2", + "sales_value": "19.99", + "retail_disc": "1.00", + "coupon_disc": "0.50", + "store_id": "st-5", + }).encode("utf-8") + + event, dlq = CustomerInteractionEvent.from_raw_payload( + payload, EventType.TRANSACTION) + self.assertIsNone(dlq) + self.assertIsNotNone(event) + self.assertEqual(event.household_key, "hh-99") + self.assertEqual(event.transaction_id, "tx-888") + self.assertEqual(event.event_type, EventType.TRANSACTION.value) + self.assertIsNotNone(event.transaction) + self.assertEqual(event.transaction.product_id, "prod-abc") + self.assertEqual(event.transaction.quantity, 2) + self.assertEqual(event.transaction.sales_value, 19.99) + self.assertEqual(event.transaction.retail_disc, 1.00) + self.assertEqual(event.transaction.coupon_disc, 0.50) + self.assertIsNone(event.coupon) + + event_dict = event.to_dict() + self.assertEqual(event_dict["household_key"], "hh-99") + self.assertIsNotNone(event_dict["transaction"]) + self.assertIsNone(event_dict["coupon"]) + + def test_customer_interaction_event_from_raw_payload_coupon(self): + payload = { + "household_key": "hh-99", + "transaction_id": "tx-888", + "coupon_upc": "cp-99999", + "campaign": "promo-2026", + "day": 42, + } + + event, dlq = CustomerInteractionEvent.from_raw_payload( + payload, EventType.COUPON) + self.assertIsNone(dlq) + self.assertIsNotNone(event) + self.assertEqual(event.household_key, "hh-99") + self.assertEqual(event.transaction_id, "tx-888") + self.assertEqual(event.event_type, EventType.COUPON.value) + self.assertIsNotNone(event.coupon) + self.assertEqual(event.coupon.coupon_upc, "cp-99999") + self.assertEqual(event.coupon.campaign, "promo-2026") + self.assertEqual(event.coupon.day, 42) + self.assertIsNone(event.transaction) + + def test_customer_interaction_event_from_raw_payload_malformed_json(self): + event, dlq = CustomerInteractionEvent.from_raw_payload( + b"NOT_A_JSON_STRING", EventType.TRANSACTION) + self.assertIsNone(event) + self.assertIsNotNone(dlq) + self.assertEqual(dlq.source, EventType.TRANSACTION.value) + self.assertIn("Malformed payload", dlq.error_message) + + def test_customer_interaction_event_from_raw_payload_missing_keys(self): + payload = json.dumps({"product_id": "prod-1"}) + event, dlq = CustomerInteractionEvent.from_raw_payload( + payload, EventType.TRANSACTION) + self.assertIsNone(event) + self.assertIsNotNone(dlq) + self.assertIn("Missing required household_key", dlq.error_message) + + def test_customer_interaction_event_from_raw_payload_unsupported_type(self): + event, dlq = CustomerInteractionEvent.from_raw_payload( + 12345, EventType.TRANSACTION) # type: ignore[arg-type] + self.assertIsNone(event) + self.assertIsNotNone(dlq) + self.assertIn("Unsupported payload type", dlq.error_message) + + def test_beam_schema_compatibility(self): + for model_cls in ( + TransactionItem, + CouponRedemption, + CustomerInteractionEvent, + UnifiedTransactionRecord, + CustomerSessionProfile, + ): + schema = named_tuple_to_schema(model_cls) + self.assertIsNotNone(schema) + self.assertGreater(len(schema.fields), 0) if __name__ == "__main__": diff --git a/pipelines/cdp/tests/test_simulator.py b/pipelines/cdp/tests/test_simulator.py new file mode 100644 index 00000000..60ffe18f --- /dev/null +++ b/pipelines/cdp/tests/test_simulator.py @@ -0,0 +1,68 @@ +# 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. +"""Unit tests for the synthetic Customer Data Platform event simulator.""" + +import unittest +from unittest.mock import MagicMock + +from simulator.generator import generate_synthetic_session_events +from simulator.publisher import get_topic_path + + +class SimulatorTest(unittest.TestCase): + """Unit tests for data generation and Pub/Sub topic path formatting.""" + + def test_generate_synthetic_session_events(self): + household_key = "hh-100" + base_tx_id = 999000 + + transactions, coupons = generate_synthetic_session_events( + household_key, base_tx_id) + + self.assertIsInstance(transactions, list) + self.assertGreater(len(transactions), 0) + for tx in transactions: + self.assertEqual(tx["household_key"], "hh-100") + self.assertEqual(tx["transaction_id"], "999000") + self.assertIn("product_id", tx) + self.assertIn("sales_value", tx) + self.assertGreater(tx["quantity"], 0) + self.assertGreater(tx["sales_value"], 0.0) + self.assertIn("event_timestamp", tx) + + self.assertIsInstance(coupons, list) + for cp in coupons: + self.assertEqual(cp["household_key"], "hh-100") + self.assertEqual(cp["transaction_id"], "999000") + self.assertIn("coupon_upc", cp) + self.assertIn("campaign", cp) + self.assertIn("event_timestamp", cp) + + def test_get_topic_path_fully_qualified(self): + mock_publisher = MagicMock() + full_path = "projects/test-proj/topics/test-topic" + res = get_topic_path(mock_publisher, "other-proj", full_path) + self.assertEqual(res, full_path) + mock_publisher.topic_path.assert_not_called() + + def test_get_topic_path_short_name(self): + mock_publisher = MagicMock() + mock_publisher.topic_path.return_value = "projects/my-proj/topics/my-topic" + res = get_topic_path(mock_publisher, "my-proj", "my-topic") + self.assertEqual(res, "projects/my-proj/topics/my-topic") + mock_publisher.topic_path.assert_called_once_with("my-proj", "my-topic") + + +if __name__ == "__main__": + unittest.main() diff --git a/terraform/cdp/README.md b/terraform/cdp/README.md index c72ed1c6..3569aace 100644 --- a/terraform/cdp/README.md +++ b/terraform/cdp/README.md @@ -15,7 +15,9 @@ The scripts will create the following application-level resources: | **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. | +| **BigQuery Table** | `unified_customer_data` | The destination BigQuery table storing granular unified transaction and coupon redemption records. | +| **BigQuery Table** | `customer_sessions` | The destination BigQuery table storing sessionized Customer 360 profile aggregations. | +| **BigQuery Table** | `cdp_deadletter` | The dead-letter BigQuery table capturing unparseable or rejected streaming 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`). | ## Configuration variables @@ -33,7 +35,9 @@ The scripts will create the following application-level resources: | `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. | +| `bq_table` | `string` | `"unified_customer_data"` | The BigQuery output table name for granular unified customer data. | +| `bq_sessions_table` | `string` | `"customer_sessions"` | The BigQuery output table name for sessionized Customer 360 profiles. | +| `bq_deadletter_table` | `string` | `"cdp_deadletter"` | The BigQuery dead-letter table name for malformed or rejected records. | ## How to deploy diff --git a/terraform/cdp/main.tf b/terraform/cdp/main.tf index a53a2520..969b29f5 100644 --- a/terraform/cdp/main.tf +++ b/terraform/cdp/main.tf @@ -13,18 +13,20 @@ # limitations under the License. locals { - 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 + 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 + bigquery_sessions_table = var.bq_sessions_table + bigquery_deadletter_table = var.bq_deadletter_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" { @@ -177,11 +179,71 @@ resource "google_bigquery_table" "unified_customer_data" { deletion_protection = !var.destroy_all_resources schema = jsonencode([ + { name = "session_id", type = "STRING", mode = "REQUIRED" }, { name = "transaction_id", type = "STRING", mode = "REQUIRED" }, - { name = "household_key", type = "STRING", mode = "NULLABLE" }, - { name = "coupon_upc", type = "STRING", mode = "NULLABLE" }, + { name = "household_key", type = "STRING", mode = "REQUIRED" }, { name = "product_id", type = "STRING", mode = "NULLABLE" }, - { name = "coupon_discount", type = "STRING", mode = "NULLABLE" } + { name = "quantity", type = "INTEGER", mode = "NULLABLE" }, + { name = "sales_value", type = "FLOAT", mode = "NULLABLE" }, + { name = "store_id", type = "STRING", mode = "NULLABLE" }, + { name = "retail_disc", type = "FLOAT", mode = "NULLABLE" }, + { name = "coupon_discount", type = "FLOAT", mode = "NULLABLE" }, + { name = "coupon_match_disc", type = "FLOAT", mode = "NULLABLE" }, + { name = "coupon_upc", type = "STRING", mode = "NULLABLE" }, + { name = "campaign", type = "STRING", mode = "NULLABLE" }, + { name = "day", type = "INTEGER", mode = "NULLABLE" }, + { name = "trans_time", type = "STRING", mode = "NULLABLE" }, + { name = "week_no", type = "INTEGER", mode = "NULLABLE" }, + { name = "event_timestamp", type = "TIMESTAMP", mode = "NULLABLE" }, + { name = "processed_timestamp", type = "TIMESTAMP", mode = "REQUIRED" } + ]) + + depends_on = [ + module.cdp_dataset + ] +} + +// BigQuery destination table for sessionized Customer 360 profiles +resource "google_bigquery_table" "customer_sessions" { + project = var.project_id + dataset_id = module.cdp_dataset.dataset_id + table_id = local.bigquery_sessions_table + deletion_protection = !var.destroy_all_resources + + schema = jsonencode([ + { name = "session_id", type = "STRING", mode = "REQUIRED" }, + { name = "household_key", type = "STRING", mode = "REQUIRED" }, + { name = "session_start", type = "TIMESTAMP", mode = "REQUIRED" }, + { name = "session_end", type = "TIMESTAMP", mode = "REQUIRED" }, + { name = "session_duration_sec", type = "INTEGER", mode = "REQUIRED" }, + { name = "total_transactions", type = "INTEGER", mode = "REQUIRED" }, + { name = "total_items_purchased", type = "INTEGER", mode = "REQUIRED" }, + { name = "total_spend", type = "FLOAT", mode = "REQUIRED" }, + { name = "total_discount", type = "FLOAT", mode = "REQUIRED" }, + { name = "coupons_redeemed_count", type = "INTEGER", mode = "REQUIRED" }, + { name = "distinct_products_count", type = "INTEGER", mode = "REQUIRED" }, + { name = "campaigns", type = "STRING", mode = "REPEATED" }, + { name = "stores_visited", type = "STRING", mode = "REPEATED" }, + { name = "processed_timestamp", type = "TIMESTAMP", mode = "REQUIRED" } + ]) + + depends_on = [ + module.cdp_dataset + ] +} + +// BigQuery dead-letter table for malformed or rejected records +resource "google_bigquery_table" "cdp_deadletter" { + project = var.project_id + dataset_id = module.cdp_dataset.dataset_id + table_id = local.bigquery_deadletter_table + deletion_protection = !var.destroy_all_resources + + schema = jsonencode([ + { name = "source", type = "STRING", mode = "REQUIRED" }, + { name = "raw_payload", type = "STRING", mode = "NULLABLE" }, + { name = "error_message", type = "STRING", mode = "REQUIRED" }, + { name = "timestamp", type = "TIMESTAMP", mode = "REQUIRED" } ]) depends_on = [ @@ -237,13 +299,17 @@ export DOCKER_IMAGE=$REGION-docker.pkg.dev/$PROJECT/$DOCKER_REPOSITORY/$IMAGE_NA export CONTAINER_URI=$DOCKER_IMAGE:$DOCKER_TAG export TRANSACTIONS_TOPIC=${module.transactions_topic.id} +export TRANSACTIONS_SUBSCRIPTION=projects/${var.project_id}/subscriptions/${local.transactions_sub} export COUPON_REDEMPTION_TOPIC=${module.coupon_redemption_topic.id} +export COUPON_REDEMPTION_SUBSCRIPTION=projects/${var.project_id}/subscriptions/${local.coupon_redemption_sub} export MAX_DATAFLOW_WORKERS=${local.max_dataflow_workers} export DISK_SIZE_GB=${local.worker_disk_size_gb} export MACHINE_TYPE=${local.machine_type} export BQ_DATASET=${module.cdp_dataset.dataset_id} export BQ_UNIFIED_TABLE=${google_bigquery_table.unified_customer_data.table_id} +export BQ_SESSIONS_TABLE=${google_bigquery_table.customer_sessions.table_id} +export BQ_DEADLETTER_TABLE=${google_bigquery_table.cdp_deadletter.table_id} export GCS_BUCKET=gs://${local.bucket_name}/assets/dataflow-solution-guide-cdp FILE } diff --git a/terraform/cdp/variables.tf b/terraform/cdp/variables.tf index bbe1b408..d517b710 100644 --- a/terraform/cdp/variables.tf +++ b/terraform/cdp/variables.tf @@ -81,3 +81,15 @@ variable "bq_table" { type = string default = "unified_customer_data" } + +variable "bq_sessions_table" { + description = "The BigQuery output table name for sessionized customer 360 profiles." + type = string + default = "customer_sessions" +} + +variable "bq_deadletter_table" { + description = "The BigQuery dead-letter table name for malformed or rejected records." + type = string + default = "cdp_deadletter" +} diff --git a/use_cases/CDP.md b/use_cases/CDP.md index 11db764b..8649eb25 100644 --- a/use_cases/CDP.md +++ b/use_cases/CDP.md @@ -1,33 +1,107 @@ # Customer Data Platform At its core, a real-time CDP is a sophisticated software solution designed to unify customer data from various sources, providing a single, comprehensive view of each individual customer. The "real-time" element is crucial: it emphasizes the ability to collect, process, and analyze customer data as events occur, enabling businesses to respond instantly to changing customer behaviors and preferences. -Real-time Customer Data Platforms represent a powerful tool for businesses seeking to create more personalized, engaging, and effective customer experiences. By centralizing customer data and enabling real-time analysis, CDPs unlock a new level of customer understanding and responsiveness, leading to better marketing outcomes and stronger customer relationships. -## Documentation +This reference architecture demonstrates how to ingest multi-stream customer events (transactions and coupon redemptions) from **Cloud Pub/Sub**, reconstruct customer journeys and shopping sessions via Apache Beam's dynamic **`Sessions(gap_size)` windowing**, aggregate Customer 360 session metrics, isolate invalid payloads into a **Dead-Letter Queue (DLQ)**, and write high-throughput records into **BigQuery** using the **Storage Write API**. -- [Real-time Customer Data Platform Solution Guide and Architecture (PDF)](./guides/cdp_dataflow_guide.pdf) +## Architecture Overview -## Assets included in this repository +```mermaid +flowchart LR + subgraph Ingestion["Ingestion"] + T1["Pub/Sub: cdp-transactions"] + T2["Pub/Sub: cdp-coupon-redemption"] + end -- [Terraform code to deploy infrastructure for Customer Data Platform](../terraform/cdp/) -- [Sample pipelines in Python for Customer Data Platform](../pipelines/cdp/) + subgraph Dataflow["Google Cloud Dataflow (Apache Beam)"] + P1["ParseRecordDoFn\n(Safe JSON + Validation)"] + DLQ_BRANCH["Dead-Letter Errors\n(Side Output)"] + SESS["Assign Timestamps &\nSessions(gap_size) Windowing"] + GBK["GroupByKey\n(by household_key)"] + PROC["ProcessCustomerSessionDoFn"] + UNIF["Unified Transactions\n(with Session ID)"] + C360["Customer 360 Session\nProfiles (Tagged Output)"] + end + + subgraph Storage["Google BigQuery (Storage Write API)"] + BQ_UNIF["unified_customer_data\n(Granular items)"] + BQ_SESS["customer_sessions\n(Customer 360 aggregates)"] + BQ_DLQ["cdp_deadletter\n(Error audit)"] + end -## Technical benefits + T1 --> P1 + T2 --> P1 + P1 -.->|errors| DLQ_BRANCH + P1 -->|valid| SESS + DLQ_BRANCH --> BQ_DLQ + SESS --> GBK --> PROC + PROC -->|main| UNIF --> BQ_UNIF + PROC -.->|sessions| C360 --> BQ_SESS +``` -Dataflow provides enormous advantages as a platform for your Customer Data Platform use -cases: +## Documentation -- **Real-Time Data Ingestion and Processing**: Dataflow enables the seamless and efficient movement of customer data from various sources into the CDP in real-time. This ensures that the CDP is always working with the most up-to-date information, allowing for timely insights and actions. +- [Real-time Customer Data Platform Solution Guide and Architecture (PDF)](./guides/cdp_dataflow_guide.pdf) -- **Enhanced Data Transformation and Enrichment**: Dataflow pipelines can perform complex transformations on incoming data, ensuring it is clean, standardized, and formatted correctly for the CDP. - Additionally, dataflow can enrich customer data with additional context or attributes from external sources, leading to more complete and valuable customer profiles. +## Assets included in this repository + +- [Terraform code to deploy infrastructure for Customer Data Platform](../terraform/cdp/) +- [Sample streaming pipeline in Python for Customer Data Platform](../pipelines/cdp/) -- **Scalability and Flexibility**: Dataflow solutions are designed to handle large volumes of data and can scale effortlessly to accommodate growing data needs. They offer flexibility in terms of data sources, processing logic, and output destinations, making them adaptable to evolving business requirements. +## Key Architectural Capabilities -- **Automation and Efficiency**: Dataflow pipelines can automate data ingestion, transformation, and delivery processes, reducing manual effort and minimizing errors. This streamlines data management, freeing up resources for more strategic tasks. +- **Dynamic Event-Time Sessionization (`window.Sessions`)**: + - Reconstructs customer shopping journeys by dynamically grouping events that occur within an inactivity gap (default 15 minutes). + - Handles late-arriving data safely with watermark-based accumulating triggers (`AccumulationMode.ACCUMULATING`) and allowed lateness windows. + - **Late Event Re-evaluation & Deduplication**: When late events (e.g. delayed coupon redemptions) arrive after the watermark passes, the accumulating trigger re-evaluates the entire session window to join late events with earlier transactions and update Customer 360 session totals. Because BigQuery sinks use `WRITE_APPEND`, late firings append updated snapshots. Downstream consumers can easily deduplicate by `session_id` or `transaction_id` using `processed_timestamp`. +- **Customer 360 Session Profile Aggregation**: + - Automatically calculates session metrics: total spend, basket size, coupons redeemed, distinct products purchased, stores visited, and campaigns engaged. +- **Production Dead-Letter Queue (DLQ)**: + - Diverts unparseable payloads or missing key violations to a dedicated BigQuery dead-letter table without halting pipeline execution. +- **High-Throughput Storage Write API**: + - Streams granular transaction items and session summaries into BigQuery using `STORAGE_WRITE_API` for immediate analytical availability. +- **Zero Public IP Security**: + - Fully compliant with enterprise networking guardrails, enforcing `--no_use_public_ip` and dedicated worker service accounts. -- **Improved Data Quality and Governance**: Dataflow enables data validation and cleansing during the ingestion process, ensuring data accuracy and consistency. Data lineage and audit capabilities within dataflow tools help track data transformations and maintain data governance standards. +## Quickstart & Verification -- **Actionable Insights and Personalization**: By feeding clean and enriched data into the CDP in real-time, dataflow enables the CDP to generate more accurate and timely insights. These insights can be used to trigger personalized marketing campaigns, recommendations, and customer interactions, leading to improved engagement and conversions. +1. **Provision Infrastructure**: + ```bash + cd terraform/cdp + terraform init && terraform apply + ``` +2. **Launch Streaming Pipeline**: + ```bash + cd ../../pipelines/cdp + source scripts/00_set_environment.sh + ./scripts/01_build_and_push_container.sh + ./scripts/02_run_dataflow.sh + ``` +3. **Generate Streaming Events**: + ```bash + python3 ./scripts/03_publish_events.py --continuous --interval=1.0 + ``` +4. **Inspect Unified Results & Customer 360 Profiles in BigQuery**: + ```bash + bq query --use_legacy_sql=false 'SELECT session_id, household_key, product_id, sales_value, coupon_upc FROM cdp_dataset.unified_customer_data LIMIT 10' + bq query --use_legacy_sql=false 'SELECT session_id, household_key, total_spend, total_transactions, coupons_redeemed_count FROM cdp_dataset.customer_sessions LIMIT 10' + ``` +5. **Deduplicate Records Downstream in BigQuery**: + When late-arriving events trigger pane updates, query the latest state using BigQuery's `QUALIFY` clause: + ```sql + -- Deduplicate Customer 360 session profiles to retrieve the latest snapshot + SELECT * + FROM `cdp_dataset.customer_sessions` + QUALIFY ROW_NUMBER() OVER ( + PARTITION BY session_id + ORDER BY processed_timestamp DESC + ) = 1; -- **Omnichannel Customer Experiences**: Dataflow supports the seamless integration of customer data across various touchpoints and channels. This allows the CDP to orchestrate consistent and personalized customer experiences across the entire customer journey. + -- Deduplicate granular unified basket items + SELECT * + FROM `cdp_dataset.unified_customer_data` + QUALIFY ROW_NUMBER() OVER ( + PARTITION BY transaction_id, product_id, COALESCE(coupon_upc, '') + ORDER BY processed_timestamp DESC + ) = 1; + ```