How to load too many small json files to databricks delta table from ADLS ?

 You often come across situation where you have many small json files (telemetry files created by IOT devices) created in your ADLS/Blob Storage in different folder hierarchies. You write normal spark dataframe load code and you start facing problems like the code takes too much time to create and load the dataframe, sometimes you face memory issues or you end up increasing clusters size or numbers. 

Loading data from millions of tiny files fails even on a scaled-up Spark cluster due to the Small File Problem, which is primarily a driver bottleneck and task overhead issue, not a raw compute limitation.

Here is why adding more cluster nodes fails to fix it:

• Driver Metadata Overload: Before processing data, the Spark driver must communicate with cloud storage (ADLS/S3) to list and catalog every single file path. If you have millions of files, this metadata listing phase single-handedly overwhelms the driver memory and network, often causing an Out-Of-Memory (OOM) error before tasks even start.

• Task Scheduling Overhead: Spark creates at least one task per file. Scheduling, serializing, and dispatching millions of small tasks to executors takes more time and network overhead than the actual data processing.

• Under-utilized Executors: Executors spend most of their time waiting for tiny tasks to initialize and close out rather than performing meaningful compute. Disk I/O becomes random and inefficient because storage systems are optimized for fewer, larger contiguous blocks (e.g., 128MB+ parquet/json files).

How to fix the issue ?

To fix this issue we need to change our code from using normal dataframe load to spark streaming and cloudfiles options, this helps to load data using many tiny batches. This you can set your batch size in options to reduce the resource used, there are options like checkpointing or schema discovery which helps if the job fails while finding new json schema, you can effectively restart the job and it continues from that point. 

How to do it ?

To efficiently ingest a high volume of JSON files distributed across nested folders into a Delta table, you should use Spark Structured Streaming with spark.readStream.format("cloudFiles") (commonly known as Auto Loader). Auto Loader automatically detects and processes new files as they arrive without requiring you to manually track directory lists, making it ideal for scaling across complex folder structures.

Here is how to set it up:

Implementation Steps

1. Configure the Auto Loader Stream: Point the reader to your root directory. Auto Loader will recursively discover files across all subfolders and maintain state using checkpointing.

   Python

   cloud_files_options = {
       "cloudFiles.format": "json",
       "cloudFiles.inferColumnTypes": "true",
       "cloudFiles.schemaLocation": "/path/to/checkpoint/schema_location"
   }
   
   streaming_df = (
       spark.readStream.format("cloudFiles")
       .options(**cloud_files_options)
       .load("dbfs:/path/to/root/source/folder/")
   )
   

2. Write to the Delta Table with Checkpointing: Stream the data into your target Delta table. The checkpoint directory ensures that the stream can recover from failures exactly where it left off, avoiding duplicate processing.

   Python

   checkpoint_path = "/path/to/checkpoint/streaming_location"
   delta_table_path = "/path/to/delta/target_table"
   
   query = (
       streaming_df.writeStream
       .format("delta")
       .outputMode("append")
       .option("checkpointLocation", checkpoint_path)
       .trigger(availableNow=True) # Or use trigger(processingTime='5 minutes') for continuous streaming
       .toTable("my_target_delta_table")
   )
   

Verification

• Check Stream Status: Run query.status or monitor your Spark UI streaming tab to verify that files are being discovered, batches are processing successfully, and the number of input rows matches expectations.

• Verify Delta Table: Run SELECT COUNT(*) FROM my_target_delta_table or inspect table history using DESCRIBE HISTORY my_target_delta_table to confirm that new data has been appended correctly.