Hi @shan-databricks , Try with below option

df = ( spark.readStream .format("cloudFiles")

.option("cloudFiles.format", "csv")

.option("badRecordsPath", "/mnt/my-bad-records")

#.option("mode", "PERMISSIVE") # Do NOT set this!

.schema(my_schema) .load("/mnt/data") )