Skip to content

fix(cdp): pass schema as runtime option and add typed join handling - #224

Merged
iht merged 2 commits into
GoogleCloudPlatform:mainfrom
adi-IL:fix/cdp-schema-loading-and-type-annotations
Aug 24, 2026
Merged

fix(cdp): pass schema as runtime option and add typed join handling#224
iht merged 2 commits into
GoogleCloudPlatform:mainfrom
adi-IL:fix/cdp-schema-loading-and-type-annotations

Conversation

@adi-IL

@adi-IL adi-IL commented Aug 23, 2026

Copy link
Copy Markdown
Contributor

Fixes #88, Fixes #89

What was wrong

  1. In pipelines/cdp/cdp_pipeline/customer_data_platform.py, the BigQuery output schema was read directly at module import time with a hardcoded relative path (with open("./schema/unified_table.json", encoding="utf-8")). When the package was installed on Dataflow workers or executed outside the pipelines/cdp directory, the relative path failed with FileNotFoundError.
  2. In left_join, coupon_redempt_value lacked proper type annotations and type checks, triggering linter warnings regarding dictionary subscriptability on Optional types.

What changed

  • Removed top-level schema file loading from module scope.
  • Added load_output_schema(schema_path) helper function that supports custom schema paths, package-relative fallback paths, and embedded default schema definitions.
  • Added --output_schema_path argument to MyPipelineOptions in pipelines/cdp/cdp_pipeline/options.py.
  • Updated create_and_run_pipeline to load and pass the schema dynamically at pipeline launch time.
  • Added complete type annotations and defensive dictionary checks to left_join for clean linter validation.
  • Added unit tests in pipelines/cdp/tests/test_customer_data_platform.py covering matching joins, non-matching joins, schema loading, and Beam pipeline transformation execution with TestPipeline.

Verification

  • yapf -i -r --style yapf . -> clean formatting
  • pylint --rcfile ../pylintrc cdp_pipeline tests -> 10.00/10 rating
  • pytest tests/ -> 4 passed
  • python setup.py sdist -> package builds successfully

@iht iht left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM! Verified that both #88 (schema loading at launch time with package fallback and CLI option) and #89 (type annotations, defensive dictionary handling in left_join, and unit tests) are fully resolved. All CI checks passed successfully.

@iht
iht merged commit 73f05d1 into GoogleCloudPlatform:main Aug 24, 2026
11 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Customer Data Platform: value is probably not a dictionary Customer Data Platform: pass schema as option to PTransform

2 participants