feat(pipelines/cdp): finalize customer data platform with sessionization and customer 360 profiles - #274
Open
iht wants to merge 4 commits into
Open
Conversation
…ion and customer 360 profiles - Implement dynamic event-time sessionization using Sessions(gap_sec) with watermark-based late-data triggers - Add ProcessCustomerSessionDoFn to produce granular items with session IDs and aggregate Customer 360 session profiles - Implement Dead-Letter Queue (DLQ) tagged side output to isolate malformed payloads and missing keys - Add high-throughput BigQuery Storage Write API dual sinks (unified_customer_data and customer_sessions) plus DLQ table (cdp_deadletter) - Enhance generate_transaction_data.py to simulate realistic multi-event customer shopping journeys with --continuous, --interval, --count, and --inject_errors - Add DirectRunner local execution script (02_run_local.sh) and update 02_run_dataflow.sh with subscription and sink parameters - Update Terraform module in terraform/cdp to provision customer_sessions and cdp_deadletter tables and dynamic environment variables - Expand test suite to 11 unit and pipeline integration tests with 100% pass rate and 10.00/10 PyLint rating - Promote Customer Data Platform solution guide from Beta to Ready status TAG=agy CONV=69d55818-9356-47fa-9e0d-cd9e5b0bfe03
…s with Beam schemas - Introduce strongly-typed NamedTuple models in cdp_pipeline/models.py with native Beam schema compatibility: TransactionItem, CouponRedemption, CustomerInteractionEvent, DeadLetterRecord, UnifiedTransactionRecord, and CustomerSessionProfile. - Use EventType(StrEnum) for event categorization without string duplication. - Refactor ParseRecordDoFn to emit composed CustomerInteractionEvent and typed deadletter side-outputs. - Refactor ProcessCustomerSessionDoFn to operate over typed NamedTuples and produce UnifiedTransactionRecord and CustomerSessionProfile. - Optimize BigQuery streaming sinks to use Storage Write API with auto-sharding and 5-second triggering frequency. - Update test suite with 100% pass rate and add comprehensive unit tests for models and Beam schema compatibility. TAG=agy CONV=69d55818-9356-47fa-9e0d-cd9e5b0bfe03
Member
Author
Update: Migrated to Composable NamedTuples with Native Beam Schemas
|
iht
commented
Sep 8, 2026
Comment on lines
+704
to
+707
| output_schema: Optional[Union[Dict[str, Any], str]] = None, | ||
| ): | ||
| """Launches the Customer Data Platform streaming pipeline on Dataflow or DirectRunner.""" | ||
| del output_schema # Handled via options or schema loader |
Member
Author
There was a problem hiding this comment.
Why is this necessary? Passing some argument that is not used and then deleting that.
Member
Author
There was a problem hiding this comment.
Good catch! You're completely right. That parameter was leftover from an older schema-passing prototype and is no longer needed since schema loading is handled internally in build_pipeline. Removed output_schema and the del statement in commit bc1db97.
…eate_and_run_pipeline - Remove unused output_schema parameter and del statement from create_and_run_pipeline. - Schema loading is dynamically handled inside build_pipeline from pipeline options and packaged schema definitions. TAG=agy CONV=69d55818-9356-47fa-9e0d-cd9e5b0bfe03
…ata simulator - Split 673-line customer_data_platform.py into cohesive, single-responsibility modules: - cdp_pipeline/schemas.py: BigQuery schema loading and defaults - cdp_pipeline/parsing.py: ParseRecordDoFn and AssignEventTimestampDoFn - cdp_pipeline/sessionization.py: ProcessCustomerSessionDoFn, legacy joins - cdp_pipeline/sinks.py: BigQuery Storage Write API sink builders - cdp_pipeline/pipeline.py: build_pipeline DAG assembly and create_and_run_pipeline - cdp_pipeline/customer_data_platform.py: Backward-compatible facade re-exporting all symbols - Extract data generation out of Beam worker package into standalone simulator: - simulator/generator.py: Synthetic shopping basket and event generation - simulator/publisher.py: Async Pub/Sub streaming publisher - scripts/03_publish_events.py: Standardized CLI launcher script - cdp_pipeline/generate_transaction_data.py: Backward-compatible shim forwarding to simulator - Add unit tests for simulator and facade export parity (24/24 tests passing). - Maintain 10.00/10 PyLint score and full compliance with Google Python style. TAG=agy CONV=69d55818-9356-47fa-9e0d-cd9e5b0bfe03
Member
Author
Update: Codebase Modularization & Simulator Extraction
|
iht
commented
Sep 8, 2026
Comment on lines
+30
to
+35
| warnings.warn( | ||
| "cdp_pipeline.generate_transaction_data has moved to " | ||
| "simulator.publisher and scripts/03_publish_events.py.", | ||
| DeprecationWarning, | ||
| stacklevel=2, | ||
| ) |
Member
Author
There was a problem hiding this comment.
Remove this, we are developing this from scratch, there is no need for backwards compatibility.
| """ | ||
| A data generator for the Customer Data Platform analytics pipeline. | ||
| """ | ||
| """Backward-compatible entry point forwarding to simulator.publisher.""" |
Member
Author
There was a problem hiding this comment.
There is no need for backwards compatibility
| import os | ||
| from typing import Any, Dict, Optional, Union | ||
|
|
||
| DEFAULT_OUTPUT_SCHEMA: Dict[str, Any] = { |
Member
Author
There was a problem hiding this comment.
The schemas are also available in JSON, are these schemas duplicated from the JSON files?
Comment on lines
+216
to
+218
| """Loads a BigQuery schema from a custom path, packaged file, or fallback dict.""" | ||
| if fallback_schema is None: | ||
| fallback_schema = DEFAULT_OUTPUT_SCHEMA |
Member
Author
There was a problem hiding this comment.
Why do we need to add this? We fully control this project. Is it necessary to duplicate the schemas in JSON and Python code?
| dataflow_options.job_name = f"customer-data-platform-{now_epoch_ms}" | ||
| custom_options: MyPipelineOptions = pipeline_options.view_as( | ||
| MyPipelineOptions) | ||
| if not custom_options.project_id and dataflow_options.project: |
Member
Author
There was a problem hiding this comment.
There should not be a custom option for project, we use the same project id as for Dataflow
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
This PR finalizes the Customer Data Platform (CDP) solution guide pipeline (
pipelines/cdp/) to 100% completion by implementing dynamic event-time sessionization and Customer 360 customer journey reconstruction.Key Changes
Dynamic Event-Time Sessionization:
Sessions(gap_size)windowing (default: 15 minutes / 900 seconds).AfterWatermark(late=AfterCount(1))) andAccumulationMode.ACCUMULATINGto safely incorporate late-arriving events.AssignEventTimestampDoFnto assign timestamps fromevent_timestampor Pub/Sub metadata.Customer 360 Session Profile Aggregation:
ProcessCustomerSessionDoFnwhich emits:sessions): Customer 360 session rollups including session duration, total spend, total items, total discounts, coupon redemption count, distinct products, stores visited, and engaged campaigns.Dead-Letter Queue (DLQ):
ParseRecordDoFnwith isolated error routing via tagged outputerrors.cdp_dataset.cdp_deadletter.BigQuery Storage Write API Dual Sinks:
unified_customer_dataandcustomer_sessions) viaSTORAGE_WRITE_API.Realistic Session Data Simulator:
generate_transaction_data.pywith multi-event session journey generation, supporting--continuous,--interval,--count, and--inject_errors.Local & Dataflow Execution:
scripts/02_run_local.shforDirectRunnerlocal execution.scripts/02_run_dataflow.shwith subscription arguments, session table, DLQ table, and Storage Write API options.Terraform Infrastructure:
terraform/cdp/withcustomer_sessionsandcdp_deadletterBigQuery tables and updated environment variable exports in00_set_environment.sh.Tests & Quality Verification:
tests/test_customer_data_platform.pyto 11 unit and pipeline integration tests (100% pass rate).pipelines/pylintrc.python setup.py sdistbundles all schema files.terraform fmt -checkandterraform validatepassing cleanly.Promotion to Ready:
Beta :factory:toReady :white_check_mark:inREADME.md, updateduse_cases/CDP.md,pipelines/cdp/README.md, and deployment runbooks.TAG=agy
CONV=69d55818-9356-47fa-9e0d-cd9e5b0bfe03