Using autoloader with multiple object types in load path

[emorgoch]

My data source is going to generate csv files for multiple objects all into the same directory that I need to load from. The files will have names in the format along the lines of <objecttype>_YYYY_MM_DD_guid.csv.gz. Each objecttype will have it's own schema, and the goal is to load those files into objecttype tables.

I'm using Autoloader with cloudfiles to ingest these files. The objective is to have the process by dynamic and not have a static set of objecttypes, being able to adjust to new objecttypes being added. In my prototype, I'm performing an initial dbutils.fs.ls to get a file list and parse our the unique objecttypes. Then I loop through each objecttype, passing them to autoloader and globbing the objecttype out. But I feel this won't scale well as the number of files increases.

Is there a better method I can use to ingest these files and have them directed to the correct tables?

[ShamenParis]

Hi @emorgoch ,

Autoloader is absolutely the best tool when we just need to ingest new files as they land in storage. However, the way you are currently doing it—listing the files with dbutils and passing that list into Autoloader—isn't going to be very efficient. It actually bypasses a lot of the built-in magic that makes Autoloader so fast!

If you really need to cherry-pick specific files using dbutils, a much better approach is to use batching combined with threading.

Here is the catch you have to watch out for: standard Python threading only runs on the driver node. It can't distribute the Python threads themselves to your worker nodes. If you aren't careful, the threading will just run locally on the driver while your powerful worker nodes sit completely idle!

To get the full leverage of both your driver and your workers, we can use the threads on the driver just to kick off the Spark jobs. When we do this (and tell Spark to share resources), the driver handles the concurrency, but Spark distributes the actual heavy lifting across all your worker nodes in parallel.

By combining batching and threading this way, you'll see a massive performance boost. You can refer to the code below to see exactly how to set it up:

import concurrent.futures

# 1. Tell Spark to use the FAIR scheduler so concurrent threads can share the worker nodes
spark.conf.set("spark.scheduler.mode", "FAIR")

# Let's say this is your massive list of files from dbutils
all_files = [f"s3://your-bucket/file_{i}.parquet" for i in range(100)]

# 2. Break the files into manageable batches (e.g., 10 files per batch)
batch_size = 10
file_batches = [all_files[i:i + batch_size] for i in range(0, len(all_files), batch_size)]

# 3. Define the function that Spark will run for each batch
def process_batch(file_list):
    # Passing the list of files directly to spark.read distributes the work to the workers!
    df = spark.read.format("parquet").load(file_list)
    
    # Do your transformations and write the data
    df.write.format("delta").mode("append").save("/target/table")
    return f"Processed a batch of {len(file_list)} files successfully."

# 4. Use threading on the driver to submit these batches in parallel
# (max_workers defines how many parallel jobs the driver will try to submit at once)
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    # Submit all the batches
    futures = [executor.submit(process_batch, batch) for batch in file_batches]
    
    # Wait for them to finish
    for future in concurrent.futures.as_completed(futures):
        print(future.result())