Handling New Columns in a Databricks Data Pipeline

[gowri_databrick]

Hi everyone,

I have a question about handling schema changes during data ingestion in Databricks.

Let’s say an e-commerce company receives customer data every day with the following columns:

Customer_id, name, city

After a few months, the source system adds a new column:

email

The existing pipeline is already running successfully with the original schema.

I’m curious about how Databricks should handle this type of change.

For this scenario:

1. How does the pipeline detect that a new column has been added?
2. How can the pipeline accept the new column without causing failures?
3. What checks should be performed before automatically accepting a schema change?

How is this normally handled in a production data pipeline?

Thanks!

[data_pulse]

@gowri_databrick 

In production, this is handled via schema evolution and Databricks Auto Loader is the good candidate for it.

1) Detection: Auto Loader tracks schema in a schemaLocation (a JSON file per stream). On each batch, it compares incoming file schema against the tracked schema. When email shows up, it doesn't silently merge it, instead throws UnknownFieldException and fails the current microbatch. That failure is the detection signal, it forces the schema location to be updated before any new data with that column is processed further.

way to do it is:

.option("cloudFiles.format", "json")
.option("cloudFiles.schemaLocation", "/checkpoints/customers/schema")
.option("cloudFiles.schemaEvolutionMode", "addNewColumns")

On restart of the pipeline (via automoatic re-try set or running it manually), the schema location now includes email, and reprocessing picks it up as a proper typed column.

2) Accepting it without failure

Depends on the mode for cloudFiles.schemaEvolutionMode:

• addNewColumns (default) : auto adds new column to schema, fails once to force the update, then proceeds clean.
• rescue : new fields go into _rescuedDataColumn instead of becoming first class columns (safer if you don't want uncontrolled schema drift). 
• failOnNewColumns:  hard stop, requires manual intervention.
• none: ignores new columns entirely (dropped or rescued if rescuedDataColumn is set).

While writing into table, if mergeSchema enabled, the write accepts the new column too. Without it, you'll hit DeltaAnalysisException (schema mismatch) even if Auto Loader read it fine. Read side evolution and Write side schema are two separate concerns.

3) Pre- Acceptance checks: Before letting email auto-merge into table blindly:

Type validation: confirm inferred type matches expectations (Auto Loader infers string, double, etc. from sample data, wrong inference silently corrupts downstream logic).
Nullability/contract check: is this column expected to be always present going forward or optional?
Rescued data audit: check _rescued_data periodically even in addNewColumns mode as it catches type-mismatched values that don't fit cleanly.
Governance Evolution step: If schema evolution is not allowed, can run a selective migration script via deployment by explicitly running ALTER TABLE ... ADD COLUMNS (email STRING). This is to code base and version control all the column changes (Good for highly controlled environment). 

Data Quality checks: DLT expectations/ DQ frameworks checks and catch during the pipeline runs for schema drifts through schema/contract definitions and can alert via notifications.

In most prod setups: bronze layer uses addNewColumns or rescue liberally (schema on read, permissive), while silver/gold layers enforce a stricter contract schema with explicit ALTER TABLE mechanism, so new columns don't propagate downstream until someone (or an automated check) decides they should.

Auto Loader Reference

[Satyasai]

HI @gowri_databrick 

How does schema evolution in spark structured streaming pipelines (particularly the add new columns case) work in Databricks, both in detecting the new column and accepting the new column?

This capability in Databricks is mainly driven by two systems: Auto Loader (for reading in source files) and Delta Lake (for writing out to storage tables).

1. How it detects that there's a new column
Databricks employs Auto Loader (cloudFiles) to read in batch or streaming file data.
Schema tracking:

Initially, when your Auto Loader is set up, it creates a persistent metadata directory (the location set by the cloudFiles.schemaLocation option), which is the "source of truth" on what your file structure should be.

When new files are ingested (your daily CSV/JSON/Parquet files with Customer_id, name, city, email), Auto Loader cross-checks the headers/structure of your micro-batch with its saved _schemas state.

Once it encounters email, it flags an unknown field which triggers an internal schema update in Auto Loader by appending email to your existing schema definition.

2. How it accepts the new column without failure

By default, Delta Lake uses Schema Enforcement (Guardrails) to protect your tables from accidental corruption, meaning you can't simply append unknown columns: you'll receive a write error.

To allow your pipeline to automatically accept the new email column, you'd have to tell Delta Lake to enable Schema Evolution.

A. In your Ingestion (Auto Loader read)

Set your Auto Loader's schema evolution mode to allow new fields:

Python

df = (spark.readStream

.format("cloudFiles")

.option("cloudFiles.format", "csv") # or json, parquet etc

.option("cloudFiles.schemaLocation", "/mnt/datalake/_checkpoints/schemas/customers")

.option("cloudFiles.schemaEvolutionMode", "addNewColumns") # Auto detects new fields

.load("/mnt/datalake/landing/customers/"))

B. In your Storage (Delta Lake write)

You'd pass in .option("mergeSchema", "true") when writing out to your target Bronze/Silver Delta table:

Python

(df.writeStream

.format("delta")

.option("mergeSchema", "true") # Tells delta to evolve your table schema

.option("checkpointLocation", "/mnt/datalake/_checkpoints/write/customers")

.table("bronze_customers"))

What happens to your historical records? The Delta table schema expands. Your existing historical rows (that were ingested before email existed) will automatically return NULL when queried for the email column.

What happens to your incoming records? Your new rows will populate email as you provide it.

3. Checks to perform before automatically accepting schema changes

Allowing fully unconstrained schema evolution in your production can lead to the ingestion of bad data or downstream breakage. Before any automated acceptance of a schema change, data pipelines usually implement one or more validation steps:

Rescued data column (_rescued_data):

Auto Loader automatically creates a _rescued_data JSON column. If an incoming email value has an incompatible data type (e.g. an array vs. a string), Databricks would route the malformed data into _rescued_data instead of failing your job or polluting your table.

Type widening checks:

Check whether the change is just an additive column (email) or a type change (e.g. Customer_id changing from INT to STRING). Adding a column is safe. Changing a type (especially from integral to string) would require Type Widening rules or explicit handling to prevent silent corruption.

Data quality constraints (Delta Expectations):

In Lakeflow / Delta Live Tables (DLT), you'd attach quality checks on essential columns:

SQL

CONSTRAINT valid_customer_id EXPECT (Customer_id IS NOT NULL) ON VIOLATION DROP ROW

This ensures you can safely let in new columns like email, while ensuring your core operational keys remain uncompromised.

4. How this is handled in production (Medallion architecture)

In an enterprise Lakehouse, schema evolution is decoupled between layers:

Source Data ──> [Bronze Layer] ──> [Silver Layer] ──> [Gold Layer]

(Permissive) (Strict/Clean) (Business Views)

Bronze Layer (Permissive Ingestion)

Rule: "Never break ingestion."

Implementation: Auto Loader ingests your raw data with addNewColumns and mergeSchema = true. Your new email column lands in the raw Bronze table without requiring any manual intervention from engineers.

Silver Layer (Controlled Evolution)

Rule: "Clean, standardize, and validate."

Implementation: Your downstream Silver tables select specific columns or automatically inherit the expanded Bronze schema. Your alert or monitoring test would flag that a new field (email) has landed in Bronze. Your analytics engineers can then add business context or PII masking rules (e.g. encrypting email) before exposing it to Gold layers.

Gold Layer (Strict Compatibility)

Rule: "Zero breaking changes for BI/Reporting."

Implementation: Your Gold models rely on either explicit column selections or semantic views (CREATE VIEW gold_customers AS SELECT customer_id, name, city FROM silver_customers). The addition of email in Bronze/Silver won't break your existing Power BI or Tableau dashboards because the Gold view explicitly serves only the columns downstream applications expect until the data team intentionally updates the view.

ADDITIONALY See This Below Link
https://community.databricks.com/t5/get-started-discussions/schema-evolution-and-schema-enforcement-...

[balajij8]

@gowri_databrick 

Auto Loader automatically detects and handles new columns when ingesting files. When your source adds the email column, Auto Loader's checkpoint tracks the schema change and can either rescue unknown columns into _rescued_data (safe default) or automatically accept them with schemaEvolutionMode => addNewColumns. Manual intervention is not required as the pipeline detects the change on the next file arrival.

Production pipelines should use a Bronze Silver Gold pattern where Bronze accepts all columns permissively, Silver validates and transforms with explicit column selection and Gold serves business-ready data. This prevents downstream breakage even if Bronze gets 10 new unexpected columns, the Silver layer explicitly selects only customer_id, name, city, email so queries don't break. Add quality checks at Silver - validate email format, flag nulls in critical columns and use data quality expectations.

Before auto accepting schema changes in production, implement a monitoring and validation checks. Set up Databricks SQL alerts to notify the team when new columns appear (You can query DESCRIBE TABLE bronze_customers periodically), validate data types match expectations, check for PII in new columns and test downstream dashboards. For highly regulated environments, use schemaEvolutionMode => rescue to quarantine unexpected columns for manual review, or failOnNewColumns to require explicit approval before accepting changes. Your pipelines can be permissive at ingestion (Bronze), strict at validation (Silver) and controlled at consumption (Gold). More details here