I have a Databricks notebook that reads csv files as a first step in an ETL pipeline. Sometimes the csv-files does not have the required schema and this causes the notebook to crash. I need to handle these errors when they occur instead of letting the entire pipeline crash.
Below is my code where I attempt to handle these errors. When a faulty csv-file is read I expect the output to be "Exception caught".
try:
newData = (spark.read
.format("csv")
.option("delimiter", "|")
.option("mode", "FAILFAST")
.option("inferSchema", "false")
.option("enforceSchema", "true")
.option("header", "True")
.schema(schema)
.load(bronzePath + "/"+ fileName + "*")
)
newData.display()
except FileReadException as e:
# Do stuff, handle exception
print("Exception caught")
But the Exception is never caught and I get the the full Exception as output.
FileReadException: Error while reading file dbfs:/mnt/datalake/bronze/<myPath>/<myFile.csv>.
Caused by: SparkException: Malformed records are detected in record parsing. Parse Mode: FAILFAST. To process malformed records as null result, try setting the option 'mode' as 'PERMISSIVE'.
Caused by: BadRecordException: org.apache.spark.sql.catalyst.csv.MalformedCSVException: Malformed CSV record
Caused by: MalformedCSVException: Malformed CSV record
Googling the issue helped me understand that it is not possible to catch scala exceptions in pyspark.
Is there some other way that I can catch this exception? What other alternatives do I have?
I need to in some way handle any faulty csv-files that are received. Using PERMISSIVE or DROPMALFORMED mode is not an option for me since I need to react and treat the faulty files.