Hi @ChrisLawford_n1,

You are correct that managed file events (cloudFiles.useManagedFileEvents = true) works by having Databricks maintain a record of file events on the external location, so when you start a new Auto Loader stream, it can replay those events rather than doing a full directory listing. However, as you have observed, when cloudFiles.includeExistingFiles is set to true (which is the default), Auto Loader still needs to perform an initial directory listing on the very first run to discover all files that existed before the file events service started tracking. This initial listing is the bottleneck you are experiencing.

Here are several approaches to improve performance in your scenario:

OPTION 1: DISABLE INCLUDE EXISTING FILES FOR SUBSEQUENT STREAMS

If you have already ingested all existing files through your first Auto Loader stream, additional streams reading from the same path do not necessarily need to re-list all existing files. You can set:

.option("cloudFiles.includeExistingFiles", "false")

This tells the new Auto Loader instance to only pick up files that arrive after the stream starts. Since managed file events are active on the external location, the new stream will immediately start receiving notifications for new files without any initial listing overhead.

If you do need historical files in the new target table, consider populating it with a one-time batch read (spark.read) from the source path or from an existing Delta table, and then switch to Auto Loader for incremental processing going forward.

OPTION 2: USE BACKFILL INTERVAL INSTEAD OF INCLUDE EXISTING FILES

Rather than having Auto Loader do a potentially slow initial listing, you can set cloudFiles.includeExistingFiles to false and instead use a periodic backfill to catch any files that may have been missed:

.option("cloudFiles.includeExistingFiles", "false")
.option("cloudFiles.backfillInterval", "1 day")

Note: cloudFiles.backfillInterval is not compatible with cloudFiles.useManagedFileEvents. So if you take this approach, you would use standard file notification mode instead of managed file events. The backfill runs asynchronously and does not block your stream processing.

OPTION 3: INCREASE FETCH PARALLELISM

The cloudFiles.fetchParallelism option controls the number of threads used when fetching messages from the queueing service. The default is 1. Increasing this can help when there is a large volume of file events to replay:

.option("cloudFiles.fetchParallelism", "8")

Note that this option is documented as not applicable when cloudFiles.useManagedFileEvents is true, so this is more relevant for standard file notification mode. If you are specifically using managed file events, this may not help with the initial listing phase.

OPTION 4: PARTITION YOUR INPUT PATHS

Instead of pointing a single Auto Loader instance at a broad top-level path, consider splitting the workload across multiple streams, each pointing at a more specific sub-path. For example, if your data is organized by date or category:

# Stream 1
spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "parquet") \
    .option("cloudFiles.useManagedFileEvents", "true") \
    .load("s3://bucket/data/year=2025/")

# Stream 2
spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "parquet") \
    .option("cloudFiles.useManagedFileEvents", "true") \
    .load("s3://bucket/data/year=2026/")

Each stream has a smaller directory tree to list, and they can run in parallel. You can write all of them to the same target Delta table. This effectively gives you the per-directory-level parallelism you are looking for.

OPTION 5: INITIAL BATCH LOAD PLUS INCREMENTAL STREAMING

For very large existing datasets, the most performant pattern is often to separate the initial load from the ongoing incremental ingestion:

1. Do a one-time batch load of all existing files:

df = spark.read.format("parquet").load("s3://bucket/data/")
df.write.format("delta").mode("overwrite").saveAsTable("catalog.schema.target_table")

2. Then start Auto Loader for incremental processing with includeExistingFiles disabled:

spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "parquet") \
    .option("cloudFiles.useManagedFileEvents", "true") \
    .option("cloudFiles.includeExistingFiles", "false") \
    .load("s3://bucket/data/") \
    .writeStream \
    .option("checkpointLocation", "/checkpoints/target_table") \
    .trigger(availableNow=True) \
    .toTable("catalog.schema.target_table")

This avoids the slow listing entirely for the streaming portion.

ADDITIONAL TIPS

- Make sure you are running Databricks Runtime 15.4 LTS or newer, as this version includes improvements that prevent waiting for full RocksDB state downloads before stream startup, which can improve overall stream initialization time.

- Run your Auto Loader streams at least once every 7 days when using file events. If more than 7 days pass between runs, the file events cache expires and Auto Loader falls back to a full directory listing.

- If you are using Trigger.AvailableNow, file discovery happens asynchronously with data processing, which can improve overall throughput during the initial catch-up phase.

For the full list of Auto Loader configuration options, see:
https://docs.databricks.com/aws/en/ingestion/cloud-object-storage/auto-loader/options.html

For production best practices:
https://docs.databricks.com/aws/en/ingestion/cloud-object-storage/auto-loader/production.html

* This reply used an agent system I built to research and draft this response based on the wide set of documentation I have available and previous memory. I personally review the draft for any obvious issues and for monitoring system reliability and update it when I detect any drift, but there is still a small chance that something is inaccurate, especially if you are experimenting with brand new features.