My apologies, I read it a little incorrect originally.

For your use case I would use COPY INTO which will only load the files you have not processed yet. You could use structured streaming to do this or the Databricks AutoLoader but those would be a little more complex.

For structured streaming you can use a ".trigger(once=True)" to use the streaming API as a batch process. You would use the checkpoint location on the write to track which files have been processed.

With AutoLoader you can use the "File Listing" option to identify which files have been used last. You will still want to use the .trigger(once=True) argument here as well.

Here are examples below on how to use the COPY INTO command:

# copy into delta by providing a file location
 
COPY INTO delta.`abfss://container@storageAccount.dfs.core.windows.net/deltaTables/target`
FROM (
  SELECT _c0::bigint key, _c1::int index, _c2 textData
  FROM 'abfss://container@storageAccount.dfs.core.windows.net/base/path'
)
FILEFORMAT = CSV
PATTERN = 'folder1/file_[a-g].csv'
 
# copy into delta by providing a table but must be an existing delta table so you create it first
 
CREATE TABLE target as 
(
 _c0 long, 
_c1 integer, 
_c2 string
)
USING DELTA
 
COPY INTO target_table
FROM 'abfss://container@storageAccount.dfs.core.windows.net/base/path'
FILEFORMAT = CSV
PATTERN = 'folder1/file_[a-g].csv'