knowledge_base: add pipeline_python_cdc example

knowledge_base/pipeline_python_cdc/.gitignore

@@ -0,0 +1 @@
+.databricks

knowledge_base/pipeline_python_cdc/README.md

@@ -0,0 +1,51 @@
+# 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=<catalog>
+   export BUNDLE_VAR_schema=<schema>
+   ```
+
+2. Deploy the bundle to dev: `databricks bundle deploy -t dev --profile <profile>`
+3. Run the job: `databricks bundle run -t dev customers_cdc_job --profile <profile>`
+4. Query the current rows: `SELECT * FROM <catalog>.<schema>.customers_scd1 ORDER BY id`
+5. Query the history rows: `SELECT * FROM <catalog>.<schema>.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 <profile>`
+7. Deploy the bundle to prod: `databricks bundle deploy -t prod --profile <profile>`
+8. Remove the dev resources: `databricks bundle destroy -t dev --profile <profile>`
+9. Remove the prod resources: `databricks bundle destroy -t prod --profile <profile>`
+10. Drop the tables: `DROP TABLE IF EXISTS <catalog>.<schema>.customers_scd1` and `DROP TABLE IF EXISTS <catalog>.<schema>.customers_scd2`
+
+| Target | Pipeline mode |
+| --- | --- |
+| `dev` | Triggered (`continuous: false`) |
+| `prod` | Continuous (`continuous: true`) |

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

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

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

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: "6"

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,
+)

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}")