To make type widening work in your current setup, you can try the following modifications:

1. Add the mergeSchema option to your read operation:

   snapshot_load_df = (
       spark.readStream
       .format("cloudFiles")
       .option("cloudFiles.format", "json")
       .option("cloudFiles.inferColumnTypes", False)
       .option("cloudFiles.includeExistingFiles", True)
       .option("pathGlobFilter", "*.json.gz")
       .option("mergeSchema", "true")  # Add this line
       .schema(spark_schema)
       .load(f"abfss://YYY@{adl_name}.dfs.core.windows.net/Snapshot/{source_system_name}/{table_name}")
       # ... rest of the code
   )