Databricks Spark - OVERWRITE mode results in old 0 B parquet files

Question

I am using Databricks to write files in OVERWRITE mode to ADLS gen2. However, when running OVERWRITE, pre-existing parquet files from previous runs are converted into 0 B files and they remain so throughout the process. When write finishes, the new file(s) are written but the older 0 B files were not removed. It does not occur every time but created such files 4 out of 10 times. The write itself succeeds but downstream reads from this location fails.

Is there a way to prevent this as it creates issues reading back downstream?

Note: The transformations are quite simple – pick 1 column from ADLS location (200K rows), do distinct (brings down to 5k rows), create new columns using some simple transformations like split, to_date(), etc. There is however a pyspark UDF that is being used to read csv files and get the first row.

DETAILS

Config

• Databricks Runtime: 9.1 LTS (includes Apache Spark 3.1.2, Scala 2.12)
• Cluster: All Purpose (interactive)

Write

Write succeeds in every run. Below is the method I'm using to write to target (header is not mandatory as it is parquet):-

distinct_paths_df.write.format('parquet').mode('overwrite').option('header',True).save(mkt_staging_write_path_2)

Read Back error

Reading back (get error due to corruped file):-

distinct_paths_df_1 = spark.read.format('parquet').option('header',True).load(mkt_staging_read_path_2)

When reading back from the location, I get an error:-

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 20.0 failed 4 times, most recent failure: Lost task 0.3 in stage 20.0 (TID 813) (10.139.64.4 executor 4): org.apache.spark.SparkException: Exception thrown in awaitResult: 

Digging into the error it points to the said parquet file:

Caused by: java.io.IOException: Could not read footer for file: FileStatus{path=dbfs:/mnt/adls/nielsen/udl/market/staging_2/part-00000-tid-1798457489032676512-ff4c9118-718a-497f-999d-b648ace95552-13340-1-c000.snappy.parquet; isDirectory=false; length=0; replication=0; blocksize=0; modification_time=0; access_time=0; owner=; group=; permission=rw-rw-rw-; isSymlink=false}

This is one of the files from older data that was converted to 0 B during the current overwrite.