Hi @SantiNath_Dey,

There are two ways you can do it. The pattern below should work. 

source_path     = "s3://your-bucket/raw/json"
schema_location = "s3://your-bucket/_schemas/my_stream"
checkpoint_path = "s3://your-bucket/_checkpoints/my_stream"
target_table    = "catalog.db.nested_events"

df = (spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaLocation", schema_location)
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")  
        .option("cloudFiles.inferColumnTypes", "true")              # infer nested structure
        .option("rescuedDataColumn", "_rescued_data")               # capture mismatches/new unexpected fields
        .load(source_path)
)

(df.writeStream
   .format("delta")
   .option("checkpointLocation", checkpoint_path)
   .option("mergeSchema", "true")   # apply evolved schema (incl. nested) to Delta table
   .toTable(target_table))

The above uses  .option("cloudFiles.schemaEvolutionMode", "addNewColumns"). With this option, when a new column appears, stream fails with UnknownFieldException, Auto Loader infers the updated schema (including the new column), writes it to schemaLocation, and you restart the stream. On restart you must use mergeSchema so the Delta table picks up the new column. However, when a column’s data type changes (e.g. INT → LONG)...the schema is not evolved. Mismatched values are set to NULL and the original values go into _rescued_data.

A modified pattern is... instead of using .option("cloudFiles.schemaEvolutionMode", "addNewColumns"), use 

.option("cloudFiles.schemaEvolutionMode", "addNewColumnsWithTypeWidening")

The difference is that this mode can deal with both new columns and type changes. For new columns, it results in the same behaviour as addNewColumns (added to the schema when the stream restarts). In addition to that it also supports widenable type changes (e.g. INT → LONG, FLOAT → DOUBLE, smaller → larger DECIMAL). Auto Loader automatically widens the column type in the schema instead of treating the new values as mismatches, so they land in the main column rather than _rescued_data. The caveat with this option is that this is in public preview and needs DBR ≥ 16.4. So start with the addNewColumns and move to addNewColumnsWithTypeWidening at later time.

Hope this helps.

If this answer resolves your question, could you mark it as “Accept as Solution”? That helps other users quickly find the correct fix.