Unzipping Large Files Using Databricks PySpark

I have a scenario, where there are two blob containers belonging to two different azure storage accounts. Both containers are mounted in a Databricks workspace. The first container cnt-input has a folder with large number of zip files (20K per day), each is appx 5GB in size. The compression ratio is 1/10, meaning that when decompressed, the result is multiple csv files with total size of appx. 50GB. Zip files are pushed to this container on a daily basis. I would like to create a pyspark notebook that:

1- Process files incrementally; detect new incoming files and leave the ones already processed.

2- Take a large number of files as a batch (e.g. 20K) and unzip it, and save the decompressed csv files in the second blob container: cnt-output.

3- The process should be performant and optimal (Databricks runtime version is 12.1, and the cluster has 8 worker nodes of type Standard_DS3_v2).

You can follow below approach.

Here, I am loading zip files, extract it and upload it to blob container using
auto loader concept.
Which loads data incrementally.

import os
import zipfile

  

def extract_zip_file(file_data):
    file_path, modificationTime,length,content = file_data
	with zipfile.ZipFile(file_path.replace("dbfs:","/dbfs"), 'r') as zip_ref:
		zip_ref.extractall("/dbfs"+output_mount_path+"csv_data")

input_mount_path = "/mnt/jgsblob/cnt-input/"
output_mount_path = "/mnt/jgsblob2/cnt-ouput/"

extract_zip_file is the function used to extract files which are read by spark
readstream as below.

query = spark.readStream.format("cloudFiles").option("cloudFiles.format", "binaryFile").option("pathGlobFilter", "*.zip")\
.load(input_mount_path+"Databricks").writeStream.option("checkpointLocation", output_mount_path+"chk_point/")\
.foreachBatch(lambda batch_df, batch_id: batch_df.foreach(extract_zip_file)).start()

In this code, I used the format as cloudFiles which invokes auto loader,
and checkpointLocation where it stores the records which are already uploaded.

Below are the zip files in container one, that is input mount path
input_mount_path = "/mnt/jgsblob/cnt-input/"

Output:
Below are files extracted in container 2, that is output mount path.
output_mount_path = "/mnt/jgsblob2/cnt-ouput/"

Here, you can see initially the cosmetics.csv and voted-kaggle-dataset are created at 11:51 and I uploaded new zip file, then at 11:55 gold.csv loaded,
even though previous zip files contain in container one.

Make sure to use the correct node type according to your input data size.
Because I have tested this for small size files.