Managing Schema Drift and Evolution in Spark Declarative Pipelines (SDP)

[Khasim_1]

Hi everyone,

I’m currently diving into the Spark Declarative Pipelines (SDP) framework and evaluating it for our upcoming migration of a high-velocity event-driven pipeline.

One area I’m trying to solidify is our strategy for Schema Evolution. In traditional DLT/Spark pipelines, we often rely on mergeSchema or explicit schema enforcement to handle upstream changes. With the declarative nature of SDP, how are others handling incoming schema drift from upstream sources?

1. Does the declarative layer offer built-in "Schema Drift" modes similar to the cloudFiles.schemaEvolutionMode we’ve been using in Auto Loader?
2. For those using SDP in production, are you finding that you need to be more explicit with your schemas in the code, or is the declarative engine handling the evolution gracefully?

Have you encountered any specific edge cases where the declarative logic didn't behave as expected during a schema change, and how did you handle it?

[Satyasai]

Hi @Khasim_1
Does SDP Offer Built-in "Schema Drift" Modes? (Underlying Auto Loader/streaming Engine Defaults)

There are inherent built-in schema evolution behaviors for both ingestion (read_files) and target table writes in Spark Structured Streaming:

1. Ingestion/Source side (read_files/Auto Loader)

When reading streams, schema evolution options are specified in the source reader rather than on the sink writer

SQL

-- In SDP SQL (Streaming Table definition)

CREATE OR REFRESH STREAMING TABLE bronze_events

AS SELECT

...

FROM STREAM read_files(

'/path/to/events',

format => 'json',

schemaEvolutionMode => 'addNewColumns' -- options: addNewColumns, rescue, failOnNewFields

)

Python

# In Python SDP, we specify the reader options in our streaming source

from pyspark import pipelines as dp

@dp.table(name="bronze_events")

def bronze_events():

return (

spark.readStream

.format("cloudFiles")

.option("cloudFiles.format", "json")

.option("cloudFiles.schemaEvolutionMode", "addNewColumns") # Or 'rescue'

.load("/path/to/events")

)

2. Declarative target schema behavior

SDP target tables (Streaming Tables) have built-in schema evolution behavior for Delta Lake / target table schemas, which is different than the conventional df.writeStream.option( mergeSchema, "true"):

Using addNewColumns evolution mode on the source reader will automatically add new columns to the target Streaming Table schema

Existing historical records will have null values filled in for any new columns that were added to the source reader schema

In Explicit Schemas vs. Graceful Handling

Most production pipelines I've worked on that require heavy ingestion have a mix of schema evolution modes across the medallion layers. Common patterns include:

Bronze (raw ingestion): Schema evolution allowed (schemaEvolutionMode => 'rescue'), ingestion of loosely-typed data or unexpected schema drift captured in a '_rescued_data' field

Avoiding downstream pipeline failures due to unexpected schema changes in upstream systems (Kafka topics/JSON blobs)

Silver/Gold (business logic layer): Schema evolution disabled + explicit schema enforcement with safe typecasting

Do not pass SELECT to downstream tables; explicitly select columns from the bronze layer table

Use TRY_CAST rather than CAST to gracefully handle schema drift attempts by upstream pipelines

[Islam_hoti · Accepted solution]

Hi,

One framing correction that might save you time: there is no separate schema drift control at the declarative layer. SDP handles orchestration and dependencies between flows. Schema evolution is still owned by the source reader and by the target table semantics, exactly as before. So the answer to your first question is that you keep using cloudFiles.schemaEvolutionMode, and it works normally inside a pipeline.

Worth knowing the default is conditional. It is addNewColumns when you do not supply a schema, and none when you do. A lot of confusion about pipelines "not evolving" traces back to someone passing an explicit schema and silently getting none.

The behaviour of addNewColumns is the edge case you asked about. The stream fails when a new column appears, and restarts with the column added. Inside Lakeflow pipelines the restart is handled for you, so it looks automatic, but the update does fail once first. On a high velocity pipeline that shows up as a latency spike and, in continuous mode, as a visible blip every time upstream adds a field. If uptime matters more than immediate availability of new columns, rescue is the better default: the schema never evolves, unexpected columns land in _rescued_data as JSON, and you promote them deliberately when you are ready.

Also note that none of the modes evolve existing data types. A widening type change upstream is not handled by addNewColumns, so use schemaHints to pin the types you actually care about instead of relying on inference to stay stable.

On the downstream side, additive changes propagate fine, but non-additive ones do not. Renames, drops and type changes on a streaming table require a full refresh, and if anything downstream reads the change data feed, a CDF query cannot span a version range where a non-additive schema change occurred. That is the failure people usually hit second, after they have already solved ingestion.

On SDP versus Lakeflow specifically, since you mentioned migration: Lakeflow runs the same open source SDP core, so the authoring model is identical, but AUTO CDC, expectations and the queryable event log are Databricks additions rather than part of SDP itself. Worth checking which of those you are depending on before assuming portability.