topic Using autoloader with multiple object types in load path in Data Engineering

Using autoloader with multiple object types in load path

emorgoch — Thu, 20 Aug 2026 19:57:10 GMT

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?

Re: Using autoloader with multiple object types in load path

ShamenParis — Thu, 20 Aug 2026 21:42:38 GMT

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

Re: Using autoloader with multiple object types in load path

adnan_alvee — Fri, 21 Aug 2026 00:53:35 GMT

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:

Reads the landing path with Auto Loader.
Selects only its filenames using pathGlobFilter.
Applies that object’s explicit schema.
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