- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
03-17-2026 07:13 AM
Thanks @Ashwin_DSA
Continue the topic, given the context discussed, when there is stateful operations in silver, the mismatch is definitely in the checkpointed state, not in table schema.
I further tested it by removing the join (which is the single stateful operations) from silver transformation, keeping the _rescued_data handling and unpivot as-is,
stream writer at silver writes correctly the dalta table with the same checkpoint location.
Summary of my test is:
- bronze: reads first buntch of .csv, which has schema1. (set to use _rescued_data)
- silver: reads from bronze, transform including unpivot, write to silver.mytable using checkpoint location mychechpoint
- bronze: put more .csv to the source location, these new .csv has extra columns.
- bronze: reads the sencond buntch of .csv , new columns are saved to _rescued_data by auto loader
- silver: run the same code as above. -- reads, transforms (including unpivot), write to silver.mytable using checkpoint location mychechpoint
- verify the new rows in silver.mytable.
=== This is to confirm the cause of the mismatch, with testing.
In production, the join is to introduce meaningful identifier, i.e., primary key, to rows in .csv. I can not remove it.
Now, what are the recommended design for such scenarios?
You mentioned:
1) Deploy the updated silver job (with _rescued_data handling, join, unpivot).
2) Run it with a new checkpointLocation so it rebuilds its state from scratch.
This is essentially the same as using cloudFiles.schemaEvolutionMode = addNewColumns, in the sense that a new checkpointLocation is required.
Given this behavior, I would consider using addNewColumns is simpler since I do not have to handing _reascued_data.
3)Keep the subset of columns that participate in stateful ops stable going forward.
Handle future schema drift only in a stateless part of the pipeline
(e.g., parse _rescued_data, unpivot, then project back to the same fixed set of stateful columns before any aggregations/joins).
In production, the stateful operation (join) can not be removed.
Summary of my understanding:
- bronze: let auto loader infer and evolve schema. -- The job will fail when autoloader finds new columns, rerun will be fine
- silver: job will fail due to checkpointed state mis-match, --> remove existing checkpoint, (optionally remove the delta), rerun.
This is what we do currently. I am hoping to find a way so the "remove existing checkpoint" could be avoid.
But, after our discussion and my testing, it seems what we are doing is the most pragmatic solution.
Any other suggestions would be appreciated. I am interested in how schema changes should be handled when using structured streaming.
P.S. regarding handling/parsing _rescued_data:
trying to dynamically adding new columns reliably requires work.