databricks-spark-declarative-pipelines

Write SQL or Python transformation files locally, upload them to the workspace, then call the create_or_update_pipeline MCP tool with the pipeline name, catalog, schema, and file paths. It creates or updates the pipeline, starts a run, and returns success status with detailed errors for iteration.

Use SQL for transformations, aggregations, joins, and standard ingestion since it is simpler to develop. Use Python with the pyspark.pipelines API when you need UDFs, external API calls, ML inference, or complex custom logic that SQL cannot express.

pyspark.pipelines (imported as dp) is the modern recommended API supporting Liquid Clustering and explicit Unity Catalog paths, while dlt is the legacy Delta Live Tables API. Both are supported, but new projects should use dp with create_auto_cdc_flow instead of dlt.apply_changes.

Use the _rescued_data column to flag records with parsing errors, then split the stream into a quarantine table for investigation and a clean table for downstream processing. Set mode to PERMISSIVE in read_files so schema changes do not break the pipeline.

Use Liquid Clustering with CLUSTER BY for nearly all new tables because it adapts to data distribution, avoids small files, and removes manual OPTIMIZE work. Keep PARTITION BY only for regulatory separation, partition-drop retention, or older Delta versions.

The error occurs when a streaming table query lacks the STREAM keyword, for example using read_files without STREAM or feeding AUTO CDC from a batch source. Rewrite the source as FROM STREAM read_files(...) so the flow is streaming.