One Script to Auto-Optimize Every Delta Table in Your Fabric Workspace

Tuning and optimizing delta tables in a lakehouse can seem like a daunting task. Pipelines can start to slow down over time without reason. Unlike traditional relational databases, where you could set an index and forget it, Delta tables introduce new challenges.

You are now dealing with an ever-growing number of parquet files stitched together by JSON files in your "_delta_log" folder and a few checkpoint files consolidating the changes every so often. The beauty of writing Spark DataFrames into Delta files makes life a breeze for complex transformations or pulling data directly from a SFTP server. However, once it is loaded into your lakehouse, your work just begins.

The JSON files in the "_delta_log" folder store historical data about your Parquet files. They record which files were added or removed with every insert, update, or delete operation. By default, every tenth commit, a checkpoint file is created to consolidate this data. In a development environment, where you are making multiple corrections and changes to the original tables, the number of files generated will multiply over time.

In this article, I will focus on several reasons that cause degrading performance and then provide you with a simple script that you can schedule to alleviate your pain.

Too many small files

As I alluded to earlier in the article, your Delta tables are made up of tons of Parquet files that grow every time you insert, update, or delete a row. Over time, these start to add up. When your user runs a query, the system has to scan through all of these files to get a result. This is especially true when you are streaming data using micro-batching.

Next, you have to consider the latency for your cloud database to talk directly to your Azure Blob Storage account or your Amazon S3 bucket. Larger files are easier to read than smaller ones.

Since checkpoint files are created every 10 commits, more of them accumulate over time, and each one carries a larger stat payload to read through.

Think of this as searching through a bunch of Lego sets. Sure, you have them clean and organized, but how would you find the Batman set in a pile of 100 sets? However, if you organized this into ten categories, you would just have to go to the correct DC Comics box.

Uneven file sizes

This next issue deals with the sizes of your parquet files. Most businesses have seasonal trends. A day after Black Friday, you would expect a huge increase in your purchases. However, the exact opposite would be true for a retailer during a blizzard with poor road conditions. Assuming you are batching and storing your data daily, your file sizes could vary drastically.

Skew is a common partitioning issue with Spark queries. Think of a four-lane interstate system. Ideally, if everyone is going 70 miles per hour, your drive time is very smooth. However, if you have a car accident ahead, one lane can be blocked and start slowing down the other lanes. Spark's parallel processing can only run as fast as its slowest partition.

Only a few large files

Although rarer, having only a few really large files can also be a problem. This would most likely occur after a large historical upload and force the query to look at a very large file to find a limited amount of information. Would you want to sift through a pile of Sports Illustrated magazines just to find the one from March 2024?

The common denominator is file size and count

In order to resolve this Goldilocks scenario of different file sizes, you can use the OPTIMIZE and VACUUM functions. The OPTIMIZE function takes all your smaller files and rebuilds them into more uniformly sized files. It then marks them for deletion in the "_delta_log" files. The VACUUM command then goes behind and removes the files marked for deletion after the allotted retention period.

How the script works

The script below will automatically OPTIMIZE and VACUUM old data. Just set it on a schedule and let it run. The beauty is that it will only run these commands if they meet certain thresholds. In this case, we only run it if the average file size is less than 100 MB and there are at least 10 files. Feel free to adjust this to your needs.

size_threshold_mb = 100

min_file_count = 10

I chose 100 MB as the starting point, since anything under that size is considered a small file.  Ideally, you would want your parquet file sizes to be between 500 MB and 1 GB.  1 GB or higher is usually considered to be too big. Feel free to raise the threshold if you want more headroom, but staying under 1 GB is still the general rule.

The first step is simply to import your libraries and set up your Spark session.

from pyspark.sql import SparkSession

from delta.tables import DeltaTable

spark = SparkSession.builder.getOrCreate()

This function is the heart of the project.  It looks at each table to find the number of parquet files and the size of those files using DESCRIBE DETAIL.  Then if the file count is greater than or equal to the “min_file_count” provided, it finds the average files size.  If the average file size is less than the threshold found in the “size_threshold_mb” parameter, it calls the OPTIMIZE function and then the VACUUM function.  One caution worth pointing out is that running VACUUM will make the deletions permanent.  Anything older than the retention period (7 days by default) is no longer available for a restore (also known as time travel).

def optimize_tables(spark, table_path, size_threshold_mb, min_file_count):

    try:

        print(f"Starting health check for: {table_path}")

        table_details = spark.sql(f"DESCRIBE DETAIL {table_path}").collect()[0]

        total_bytes = table_details['sizeInBytes']

        num_files = table_details['numFiles']

        if num_files == 0:

            print('Skipping tables with zero files.\n')

            return

        avg_file_size_mb = (total_bytes / (1024 * 1024)) / num_files

        print(f"Total files: {num_files} \n")

        print(f"Average file size: {avg_file_size_mb} \n")

        need_compaction = avg_file_size_mb < size_threshold_mb

        has_enough_files = num_files >= min_file_count

        if need_compaction and has_enough_files:

            spark.sql(f"OPTIMIZE {table_path}")

            print('Optimize Complete')

            spark.sql(f"VACUUM {table_path}")

            print('Vacuum Complete')

        else:

            print(f"No need to optimize: {table_path} \n")

    except Exception as e:

        print(f"Error: {e} \n")

This is where you set your parameters: size_threshold_mb and min_file_count.  The SHOW SCHEMAS command is used to find all schemas in your lakehouse.  Each schema and table is then pulled from the schema_df and the table sent individually to the optimize_tables function from before.

size_threshold_mb = 100

min_file_count = 10

schema_df = spark.sql("SHOW SCHEMAS").collect()

for row in schema_df:

    tables = spark.sql(f"SHOW TABLES IN {row[0]}").collect()

    for table in tables:

        workspace = table[0].split('.')[0]

        database = table[0].split('.')[1]

        table_name = table[1]

        table_path = f"{workspace}.{database}.{table_name}"

        optimize_tables(

            spark=spark,

            table_path=table_path,

            size_threshold_mb= size_threshold_mb,

            min_file_count= min_file_count

        )

Once you have added this into a notebook, you can add that notebook into Data Factory and schedule it.  I have found that running this process weekly on a Sunday night works pretty well for standard batch processing. For streaming workloads, you may need to run this once or twice a day.  All of this will depend on the number of changes made.  Optimizations for streaming jobs occur more often due to the sheer number of changes.

One other key pointer is to use Z-ordering to sort your tables by the most commonly used columns. It's conceptually similar to a clustered index in relational databases. Ideally, you should focus on three or four columns, or this could actually hurt performance. I did not add this to the script since it requires more manual testing.

In conclusion, running this one simple script can optimize your Delta files in Fabric. If you are using Databricks, simply set up Predictive Optimization and Databricks will do the job for you. Now there is no need to stress over poor query performance, and you can focus on the more important issues.