I am having an issue with corrupt records in databricks. We want to count corrupt records and save corrupt records in a specific location as a delta table.
To do that we are reading using PERMISSIVE and doing queries based on this _corrupt_record column.
We are using pyspark with Apache Spark 3.0.1 in Azure Databricks .
Here the error message we get :
Since Spark 2.3, the queries from raw JSON/CSV files are disallowed when the referenced columns only include the internal corrupt record column (named _corrupt_record by default). For example: spark.read.schema(schema).json(file).filter($"_corrupt_record".isNotNull).count() and spark.read.schema(schema).json(file).select("_corrupt_record").show().
According to this documentation, you have to cache or save the data if you want to query the column corrupt records.
But we don't want to cache the data in our ETL. The ETL is used for many jobs running on the same cluster, and we can have big files of 150GB as input. Caching the data might cause the cluster to crach.
Is there any way to query these corrupt-records without caching the data ?
#1 Saving the data on a blob storage might be another option, but this sounds way to much overhead.
#2 We also tried using the option BadRecordsPath : save bad records to BadRecordsPath and read it back in order to count it, but there is no simple way to know if a file of bad records have been written (and in which partition the file has been written).
Partition looks like /20210425T102409/bad_records
See my other question about that here
#3 Another way of doing this would be to do substract dropmalformed read from permissive read. eg :
dataframe_with_corrupt = spark.read.format('csv').option("mode", "PERMISSIVE").load(path)
dataframe_without_corrupt = spark.read.format('csv').option("mode", "DROPMALFORMED").load(path)
corrupt_df = dataframe_with_corrupt.exceptAll(dataframe_without_corrupt)
But i'm not sure it would take less memory than caching!
Any suggestion or observation would be much appreciated ! Thanks in advance