From 7ffc677dc1c502ec0a3497ebf468de10bdc478f9 Mon Sep 17 00:00:00 2001 From: Andres Garcia Date: Thu, 22 Oct 2026 00:09:41 +0000 Subject: [PATCH 1/3] knowledge_base: add pipeline_python_cdc example Serverless Python pipeline with create_auto_cdc_flow (pyspark.pipelines) that writes SCD Type 1 and SCD Type 2 targets from a generated change feed. Dev is triggered, prod is continuous. A job exposes full_refresh as a job parameter. Co-authored-by: omnigent --- knowledge_base/pipeline_python_cdc/.gitignore | 1 + knowledge_base/pipeline_python_cdc/README.md | 50 +++++++++++++++++++ .../pipeline_python_cdc/databricks.yml | 37 ++++++++++++++ .../resources/changes.volume.yml | 7 +++ .../resources/customers_cdc.pipeline.yml | 16 ++++++ .../resources/customers_cdc_job.job.yml | 48 ++++++++++++++++++ .../pipeline_python_cdc/src/customers_cdc.py | 41 +++++++++++++++ .../src/generate_changes.py | 28 +++++++++++ 8 files changed, 228 insertions(+) create mode 100644 knowledge_base/pipeline_python_cdc/.gitignore create mode 100644 knowledge_base/pipeline_python_cdc/README.md create mode 100644 knowledge_base/pipeline_python_cdc/databricks.yml create mode 100644 knowledge_base/pipeline_python_cdc/resources/changes.volume.yml create mode 100644 knowledge_base/pipeline_python_cdc/resources/customers_cdc.pipeline.yml create mode 100644 knowledge_base/pipeline_python_cdc/resources/customers_cdc_job.job.yml create mode 100644 knowledge_base/pipeline_python_cdc/src/customers_cdc.py create mode 100644 knowledge_base/pipeline_python_cdc/src/generate_changes.py diff --git a/knowledge_base/pipeline_python_cdc/.gitignore b/knowledge_base/pipeline_python_cdc/.gitignore new file mode 100644 index 00000000..15bcc6dd --- /dev/null +++ b/knowledge_base/pipeline_python_cdc/.gitignore @@ -0,0 +1 @@ +.databricks diff --git a/knowledge_base/pipeline_python_cdc/README.md b/knowledge_base/pipeline_python_cdc/README.md new file mode 100644 index 00000000..54f62caf --- /dev/null +++ b/knowledge_base/pipeline_python_cdc/README.md @@ -0,0 +1,50 @@ +# Python pipeline with AUTO CDC (SCD Type 1 and Type 2) + +This Declarative Automation Bundles example defines a serverless Python pipeline that uses `create_auto_cdc_flow`. +The pipeline applies a small change feed to an SCD Type 1 table and an SCD Type 2 table. +A job writes the change feed and then refreshes the pipeline. + +For more information, see the [AUTO CDC documentation](https://docs.databricks.com/aws/en/ldp/cdc). + +## Prerequisites + +* Databricks CLI v1.20.0 or above +* A workspace with Unity Catalog and serverless compute. +* An existing catalog and schema for the tables and the `changes` volume. + +## Usage + +1. Set the catalog and the schema: + + ```sh + export BUNDLE_VAR_catalog= + export BUNDLE_VAR_schema= + ``` + +2. Deploy the bundle to dev: `databricks bundle deploy -t dev --profile ` +3. Run the job: `databricks bundle run -t dev customers_cdc_job --profile ` +4. Query the current rows: `SELECT * FROM ..customers_scd1 ORDER BY id` +5. Query the history rows: `SELECT * FROM ..customers_scd2 ORDER BY id, __START_AT` + + The late `Boston` update (seq 4) does not replace `Portland` (seq 5) in SCD Type 1. + SCD Type 2 keeps it as a closed history row: + + ``` + id name city __START_AT __END_AT + 1 Alice Seattle 1 4 + 1 Alice Boston 4 5 + 1 Alice Portland 5 null + 2 Bob Austin 2 6 + 3 Carol Denver 3 7 + 3 Carol Miami 7 null + ``` + +6. Recompute all tables: `databricks bundle run -t dev customers_cdc_job --params full_refresh=true --profile ` +7. Deploy the bundle to prod: `databricks bundle deploy -t prod --profile ` +8. Remove the dev resources: `databricks bundle destroy -t dev --profile ` +9. Remove the prod resources: `databricks bundle destroy -t prod --profile ` + +| Target | Pipeline mode | +| --- | --- | +| `dev` | Triggered (`continuous: false`) | +| `prod` | Continuous (`continuous: true`) | diff --git a/knowledge_base/pipeline_python_cdc/databricks.yml b/knowledge_base/pipeline_python_cdc/databricks.yml new file mode 100644 index 00000000..d1c3094a --- /dev/null +++ b/knowledge_base/pipeline_python_cdc/databricks.yml @@ -0,0 +1,37 @@ +# This is a Declarative Automation Bundle definition for pipeline_python_cdc. +# See https://docs.databricks.com/dev-tools/bundles/index.html for documentation. +bundle: + name: pipeline_python_cdc + +include: + - resources/*.yml + +variables: + catalog: + description: Existing Unity Catalog catalog for the pipeline tables + default: main + schema: + description: Existing schema in the catalog for the pipeline tables + default: pipeline_python_cdc + continuous: + description: Run the pipeline continuously (true) or on trigger (false) + default: false + +# The workspace host is not set here. Use a profile (--profile / -p) or the +# DATABRICKS_HOST environment variable. + +targets: + dev: + # Development mode prefixes resource names and pauses schedules. + mode: development + default: true + workspace: + root_path: /Workspace/bundle-examples/${bundle.name}/${bundle.target}/${workspace.current_user.short_name} + + prod: + mode: production + workspace: + root_path: /Workspace/bundle-examples/${bundle.name}/${bundle.target} + variables: + # Prod processes new changes as soon as they arrive. + continuous: true diff --git a/knowledge_base/pipeline_python_cdc/resources/changes.volume.yml b/knowledge_base/pipeline_python_cdc/resources/changes.volume.yml new file mode 100644 index 00000000..809c1a61 --- /dev/null +++ b/knowledge_base/pipeline_python_cdc/resources/changes.volume.yml @@ -0,0 +1,7 @@ +resources: + volumes: + # Landing zone for the generated change feed (JSON files). + changes: + catalog_name: ${var.catalog} + schema_name: ${var.schema} + name: changes diff --git a/knowledge_base/pipeline_python_cdc/resources/customers_cdc.pipeline.yml b/knowledge_base/pipeline_python_cdc/resources/customers_cdc.pipeline.yml new file mode 100644 index 00000000..aae4ac20 --- /dev/null +++ b/knowledge_base/pipeline_python_cdc/resources/customers_cdc.pipeline.yml @@ -0,0 +1,16 @@ +resources: + pipelines: + customers_cdc: + name: customers_cdc + catalog: ${var.catalog} + schema: ${var.schema} + serverless: true + continuous: ${var.continuous} + + # Read in the pipeline code with spark.conf.get("source_path"). + configuration: + source_path: ${resources.volumes.changes.volume_path}/customers + + libraries: + - file: + path: ../src/customers_cdc.py diff --git a/knowledge_base/pipeline_python_cdc/resources/customers_cdc_job.job.yml b/knowledge_base/pipeline_python_cdc/resources/customers_cdc_job.job.yml new file mode 100644 index 00000000..2810b05d --- /dev/null +++ b/knowledge_base/pipeline_python_cdc/resources/customers_cdc_job.job.yml @@ -0,0 +1,48 @@ +resources: + jobs: + customers_cdc_job: + name: customers_cdc_job + + parameters: + # Set to "true" to recompute all tables from the full change feed. + - name: full_refresh + default: "false" + + tasks: + # Write a small change feed: inserts, updates, a delete, and an out-of-order update. + - task_key: generate_changes + environment_key: default + spark_python_task: + python_file: ../src/generate_changes.py + parameters: + - ${resources.pipelines.customers_cdc.configuration.source_path} + + # pipeline_task.full_refresh takes a literal bool, so branch on the job parameter. + - task_key: check_full_refresh + depends_on: + - task_key: generate_changes + condition_task: + op: EQUAL_TO + left: "{{job.parameters.full_refresh}}" + right: "true" + + - task_key: full_refresh + depends_on: + - task_key: check_full_refresh + outcome: "true" + pipeline_task: + pipeline_id: ${resources.pipelines.customers_cdc.id} + full_refresh: true + + - task_key: refresh + depends_on: + - task_key: check_full_refresh + outcome: "false" + pipeline_task: + pipeline_id: ${resources.pipelines.customers_cdc.id} + full_refresh: false + + environments: + - environment_key: default + spec: + environment_version: "4" diff --git a/knowledge_base/pipeline_python_cdc/src/customers_cdc.py b/knowledge_base/pipeline_python_cdc/src/customers_cdc.py new file mode 100644 index 00000000..e123a382 --- /dev/null +++ b/knowledge_base/pipeline_python_cdc/src/customers_cdc.py @@ -0,0 +1,41 @@ +from pyspark import pipelines as dp +from pyspark.sql.functions import col, expr + +source_path = spark.conf.get("source_path") + + +@dp.temporary_view +def customers_changes(): + """Read the change feed incrementally with Auto Loader.""" + return ( + spark.readStream.format("cloudFiles") + .option("cloudFiles.format", "json") + .option("cloudFiles.maxFilesPerTrigger", 1) + .schema("id INT, name STRING, city STRING, operation STRING, seq INT") + .load(source_path) + ) + + +# SCD Type 1: one row per key with the latest values. Deleted keys are removed. +dp.create_streaming_table("customers_scd1") +dp.create_auto_cdc_flow( + target="customers_scd1", + source="customers_changes", + keys=["id"], + sequence_by=col("seq"), + apply_as_deletes=expr("operation = 'DELETE'"), + except_column_list=["operation", "seq"], + stored_as_scd_type=1, +) + +# SCD Type 2: full history with __START_AT and __END_AT columns. +dp.create_streaming_table("customers_scd2") +dp.create_auto_cdc_flow( + target="customers_scd2", + source="customers_changes", + keys=["id"], + sequence_by=col("seq"), + apply_as_deletes=expr("operation = 'DELETE'"), + except_column_list=["operation", "seq"], + stored_as_scd_type=2, +) diff --git a/knowledge_base/pipeline_python_cdc/src/generate_changes.py b/knowledge_base/pipeline_python_cdc/src/generate_changes.py new file mode 100644 index 00000000..40ef6079 --- /dev/null +++ b/knowledge_base/pipeline_python_cdc/src/generate_changes.py @@ -0,0 +1,28 @@ +"""Write a small customer change feed as JSON files to a Unity Catalog volume.""" + +import json +import os +import sys + +# Each file is one batch. The pipeline reads one file per micro-batch. +BATCHES = { + "batch_1.json": [ + {"id": 1, "name": "Alice", "city": "Seattle", "operation": "INSERT", "seq": 1}, + {"id": 2, "name": "Bob", "city": "Austin", "operation": "INSERT", "seq": 2}, + {"id": 3, "name": "Carol", "city": "Denver", "operation": "INSERT", "seq": 3}, + {"id": 1, "name": "Alice", "city": "Portland", "operation": "UPDATE", "seq": 5}, + ], + "batch_2.json": [ + # Late event: seq 4 arrives after seq 5 for the same key. + {"id": 1, "name": "Alice", "city": "Boston", "operation": "UPDATE", "seq": 4}, + {"id": 2, "name": None, "city": None, "operation": "DELETE", "seq": 6}, + {"id": 3, "name": "Carol", "city": "Miami", "operation": "UPDATE", "seq": 7}, + ], +} + +path = sys.argv[1] +os.makedirs(path, exist_ok=True) +for file_name, rows in BATCHES.items(): + with open(os.path.join(path, file_name), "w") as f: + f.write("\n".join(json.dumps(row) for row in rows)) + print(f"Wrote {len(rows)} changes to {path}/{file_name}") From 3666c6e1ffa76cc6cb9ee4da555177a4bd3ca3f7 Mon Sep 17 00:00:00 2001 From: Andres Garcia Date: Thu, 22 Oct 2026 00:15:01 +0000 Subject: [PATCH 2/3] Add table cleanup step to CDC README Co-authored-by: omnigent --- knowledge_base/pipeline_python_cdc/README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/knowledge_base/pipeline_python_cdc/README.md b/knowledge_base/pipeline_python_cdc/README.md index 54f62caf..e8e00be8 100644 --- a/knowledge_base/pipeline_python_cdc/README.md +++ b/knowledge_base/pipeline_python_cdc/README.md @@ -43,6 +43,7 @@ For more information, see the [AUTO CDC documentation](https://docs.databricks.c 7. Deploy the bundle to prod: `databricks bundle deploy -t prod --profile ` 8. Remove the dev resources: `databricks bundle destroy -t dev --profile ` 9. Remove the prod resources: `databricks bundle destroy -t prod --profile ` +10. Drop the tables: `DROP TABLE IF EXISTS ..customers_scd1` and `DROP TABLE IF EXISTS ..customers_scd2` | Target | Pipeline mode | | --- | --- | From 230505be4c406782f848be300dde9eff43e46ec7 Mon Sep 17 00:00:00 2001 From: Andres Garcia Date: Thu, 22 Oct 2026 02:37:43 +0000 Subject: [PATCH 3/3] Use serverless environment version 6 Move the serverless job environment from version 4 to version 6, the newest GA version. Co-authored-by: Isaac Co-authored-by: omnigent --- .../pipeline_python_cdc/resources/customers_cdc_job.job.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/knowledge_base/pipeline_python_cdc/resources/customers_cdc_job.job.yml b/knowledge_base/pipeline_python_cdc/resources/customers_cdc_job.job.yml index 2810b05d..bca1512d 100644 --- a/knowledge_base/pipeline_python_cdc/resources/customers_cdc_job.job.yml +++ b/knowledge_base/pipeline_python_cdc/resources/customers_cdc_job.job.yml @@ -45,4 +45,4 @@ resources: environments: - environment_key: default spec: - environment_version: "4" + environment_version: "6"