Catching spark exceptions in PySpark

Viewed 29

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.

0 Answers
Related