diff --git a/knowledge_base/pipeline_python_cdc/.gitignore b/knowledge_base/pipeline_python_cdc/.gitignore new file mode 100644 index 0000000..15bcc6d --- /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 0000000..e8e00be --- /dev/null +++ b/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= + 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 ` +10. Drop the tables: `DROP TABLE IF EXISTS ..customers_scd1` and `DROP TABLE IF EXISTS ..customers_scd2` + +| 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 0000000..d1c3094 --- /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 0000000..809c1a6 --- /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 0000000..aae4ac2 --- /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 0000000..bca1512 --- /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: "6" 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 0000000..e123a38 --- /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 0000000..40ef607 --- /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}")