Databricks Community

pharma70 · 3 weeks ago

Hi all,

Sometimes an upstream source adds a new column without letting us know. We do not want the Databricks workflow to fail when this happens, but we would like to be notified so we can review the new column and decide whether to include it in our processing later.

Is there a Databricks-native way to detect and alert on schema changes, while allowing the pipeline to continue running normally?

Satyasai · 3 weeks ago

Write a Databrciks Alerts with following sql server Query

SELECT
event_time,
user_identity.email AS triggered_by,
request_params.table_full_name AS table_name,
action_name,
request_params
FROM system.access.audit
WHERE action_name = 'addColumns'
AND request_params.table_full_name = 'catalog.schema.bronze_table'
AND event_time >= NOW() - INTERVAL 1 DAY;

Navigate to SQL editor -> Alerts -> Create Alert.

Set alert conditions to trigger when query result has Rows > 0.

Configure alert notifications to be delivered via Email, Slack, Microsoft Teams or Webhook/PagerDuty.

suryaprayaga · 3 weeks ago

This is a good question. Databricks Auto Loader specifically supports a "rescue schema-evolution" mode where new columns don't cause the stream to fail; unexpected fields are placed into a rescued-data column.

Besides we could safely introduce what is called as "Schema Change Detection Layer" that detects what changed between the incoming data stream vs the expected / approved schema. Take a look at a simple scenario like this:

(Expected Schema - Incoming Schema) --> New columns, Removed Columns, Schema changes, Renamed columns.

We could simply store such results in an audit table something with the following structure:

Though this capture is per-se not native within Datbaricks (others can correct me), you can perhaps use a native Databricks ability to do notification via Slack or Pager Duty.

Thanks

SP

suryaprayaga

ivanvyd · 3 weeks ago

@pharma70 if this ingestion is using Auto Loader, cloudFiles.schemaEvolutionMode = "rescue" seems like a good fit here. In that mode Auto Loader keeps the stream running, does not evolve the schema, and puts new fields into _rescued_data.

You can then alert on recent bronze rows where _rescued_data IS NOT NULL. If this runs through Lakeflow Jobs, the alert can be added as a SQL Alert task after ingestion. One useful detail is that a triggered alert still reports the task as Succeeded; the task only fails if alert evaluation itself errors.

_rescued_data also captures type and case mismatches, so for a strict "new column" notification I'd filter the rescued fields against the approved schema first.

Ivan Vydrin
Lead Software & AI Engineer · Tech Fabric LLC

Louis_Frolio · Friday

Greetings @pharma70, I did some digging and here is what I found. The replies from @Satyasai, @suryaprayaga, and @ivanvyd each cover a piece of this, so I'll stitch them together.

The right approach depends on how data arrives in Databricks.

If you ingest files with Auto Loader

cloudFiles.schemaEvolutionMode = "rescue" is built for your exact requirement, as @ivanvyd and @suryaprayaga said. One detail that surprises people: Auto Loader's default mode (addNewColumns) does fail the stream when a new column appears (an UnknownFieldException) and relies on a job retry to continue. Rescue mode flips that. The schema never changes, the stream never stops, and anything that doesn't fit lands in _rescued_data as a JSON string. Nothing is lost and downstream code keeps working.

I'd pair it with an explicit approved schema so you control exactly what's "expected":

df = (spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .schema(expected_schema)
  .option("cloudFiles.schemaEvolutionMode", "rescue")
  .option("rescuedDataColumn", "_rescued_data")
  .option("cloudFiles.schemaLocation", "/Volumes/cat/schema/_checkpoints/bronze")
  .load("/Volumes/cat/landing/source_a/"))

For the notification, a plain _rescued_data IS NOT NULL check tells you something drifted, but as Ivan noted it also fires on type and case mismatches. If you want the alert to say which new columns showed up, pull the keys out of the JSON (skipping the _file_path key Auto Loader adds):

SELECT key AS new_column, COUNT(*) AS row_count, MIN(ingest_ts) AS first_seen
FROM (
  SELECT explode(map_keys(from_json(_rescued_data, 'MAP<STRING,STRING>'))) AS key,
         ingest_ts
  FROM cat.bronze.source_a
  WHERE _rescued_data IS NOT NULL
    AND ingest_ts >= current_timestamp() - INTERVAL 1 DAY
)
WHERE key <> '_file_path'
GROUP BY key
ORDER BY first_seen;

Save that as a Databricks SQL alert (condition: row count > 0) and route it to email, Slack, Teams, or a webhook. Add it as a SQL alert task right after ingestion in your Lakeflow Job; the task reports Succeeded whether or not the alert fires, so the workflow stays green while you still get told. Keep the time window tight to your run cadence, or log notified columns to a small table, so the same rescued rows don't alert you every run. If you'd rather classify changes (new, removed, renamed, type-changed), the schema-diff audit table @suryaprayaga described is a good step up from this.

Once you've reviewed a column and want it, add it to expected_schema (or to cloudFiles.schemaHints) and it moves out of _rescued_data on the next run.

If the upstream team writes directly to a Delta table

Here the schema has already evolved by the time you see it, so @Satyasai's approach fits: look at what happened to the table. DESCRIBE HISTORY shows an ADD COLUMNS operation with the column names in operationParameters, and system.access.audit records the same addColumns event account-wide if you have system table access. Either can feed the same SQL alert pattern. Just note this detects DDL on Delta or Unity Catalog tables; it won't see a new field in raw JSON or CSV files, which is what rescue mode is for.

References