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())

[adnan_alvee]

Use one Auto Loader streaming table per object type, generated from a governed configuration registry;  not dbutils.fs.ls. Each object type needs an independent schema and checkpoint. Auto Loader cannot use one CSV stream to infer unrelated schemas and dynamically choose target tables: CSV inference produces one global schema for the stream.

Prefer changing the landing layout to:

<landing>/object_type=customer/...
<landing>/object_type=order/...

If that is impossible, keep the shared directory and assign each stream a non-overlapping pathGlobFilter.

Maintain a small registry containing:

• Validated object type
• Expected CSV schema
• Target table
• Optional CSV parsing options

Lakeflow Spark Declarative Pipelines can read this registry during pipeline planning and generate multiple streaming tables programmatically. The definitions are evaluated serially, but the resulting flows can execute in parallel. New objects become active on the next pipeline update. The registry should remain additive because removing a generated dataset causes that dataset to be dropped from the pipeline target schema.

Each generated flow:

1. Reads the landing path with Auto Loader.
2. Selects only its filenames using pathGlobFilter.
3. Applies that object’s explicit schema.
4. Writes to its own streaming table with independently managed state.

Auto Loader supports pre-compressed CSV files, including gzip. Lakeflow manages checkpoint and schema locations automatically; standalone streams require unique durable locations for each workload.

For production discovery, enable managed file events. They share one notification queue per external location, require Unity Catalog and DBR 14.3 LTS or later, and should run at least every seven days to avoid falling back to directory listing.

Minimal implementation example

import re
from pyspark import pipelines as dp

SOURCE = "/Volumes/<source_catalog>/<source_schema>/<landing_volume>"

# Small governed registry:
# object_type STRING, schema_ddl STRING, enabled BOOLEAN
objects = (
    spark.table("<config_catalog>.<config_schema>.ingestion_objects")
         .where("enabled = true")
         .select("object_type", "schema_ddl")
         .collect()
)

def define_object_table(object_type: str, schema_ddl: str):
    # Prevent target-name or glob injection.
    if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_]*", object_type):
        raise ValueError(f"Invalid object_type: {object_type}")

    @DP.table(name=f"bronze_{object_type.lower()}")
    def ingest_object():
        return (
            spark.readStream
                 .format("cloudFiles")
                 .option("cloudFiles.format", "csv")
                 .option("header", "true")
                 .option(
                     "pathGlobFilter",
                     f"{object_type}_????_??_??_*.csv.gz"
                 )
                 .option("cloudFiles.useManagedFileEvents", "true")
                 .option("rescuedDataColumn", "_rescued_data")
                 .schema(schema_ddl)
                 .load(SOURCE)
                 .selectExpr(
                     "*",
                     "_metadata.file_path AS _source_file",
                     "current_timestamp() AS _ingested_at"
                 )
        )

for row in objects:
    define_object_table(row.object_type, row.schema_ddl)

Another alternative could be to look into DLT-Meta, which is a datbaricks labs project. Its a meta data driven pipeline config for spark declarative pipelines. https://docs.databricks.com/aws/en/ldp/developer/dlt-meta

[emorgoch]

Thanks Adnan,

What I'm seeing with your solution, however, is that I need to maintain a static list of object types that I will be importing. This is what I am attempting to avoid, as system owners may expand the object types exported from the source to be ingested into Databricks without properly informing the data team.

[comb8342]

I’d avoid doing a dbutils.fs.ls followed by one Auto Loader stream per object type. That creates more streams/checkpoints as the number of object types grows.

A cleaner pattern is to use one Auto Loader stream over the directory, extract objecttype from the filename, and route each micro-batch to the appropriate Delta table with foreachBatch. Auto Loader is designed to scale to very large file counts, and the discovered file metadata is persisted in its checkpoint.

Conceptually:

df = (
spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "csv")
.option("cloudFiles.schemaLocation", schema_path)
.option("header", "true")
.load(source_path)
)

df = df.withColumn(
"objecttype",
regexp_extract(input_file_name(),
r"/([^/_]+)_\d{4}_\d{2}_\d{2}_", 1)
)

def route_batch(batch_df, batch_id):
for obj in [r.objecttype for r in batch_df.select("objecttype").distinct().collect()]:
(
batch_df.filter(col("objecttype") == obj)
.drop("objecttype")
.write
.mode("append")
.saveAsTable(f"target.{obj}")
)

df.writeStream.foreachBatch(route_batch)...

The important part is that you’re not repeatedly scanning the directory to discover files. Auto Loader handles incremental discovery, while your routing logic determines the destination table.

One caveat: because each object type has its own schema, I’d maintain a small schema/metadata registry for object type → target table/schema rather than relying entirely on inference. Auto Loader supports schema inference/evolution, but different schemas going into different tables still need deliberate handling.

That architecture should scale much better than creating a separate Auto Loader stream for every object type. stickmanhook.com.br can be a separate reference while working on the pipeline.

[emorgoch]

Thanks comb for the suggestion and the sample code. Looking into the foreachbatch processing, this seems like if could work. My one concern with it is the information here about foreachbatch serializing the processing and removing parallelization, but this would probably be acceptable given the expected data sizes of the batched exports (a few dozen files, under 100000 records a batch).