Fabric Notebooks. Get latest file from bronze Data Lake into Dataframe

I did a test which seems to work:

In my Bronze lakehouse, I created a folder which receives a new csv file every 10 minutes. 

I also created a table in my Bronze lakehouse, called processed_files_log, where I keep a log of the csv files which have already been processed and loaded into a table in my Silver Lakehouse.

I have pasted the Notebook code below. The notebook runs every 30 minutes to process new files and load the content to the Silver table. The notebook also enters the processed files' filenames into the log table.

ChatGPT helped me with creating this code, and I don't claim to fully understand all the details of the code. 

Step 1 is to list the files in the folder

# abfss path to the folder where the csv files are located 
files_path = '<insert abfss path to the folder which keeps the csv files>'

# the mssparkutils.fs.ls method lists all the files (and/or subfolders) inside the folder. I only have csv files in the folder.
files = mssparkutils.fs.ls(files_path)

# Convert FileInfo objects to list of tuples (this creates a list of the file names in the folder.)
file_data = [(file.name,) for file in files]

# This creates a dataframe consisting of the file names in the folder
df_files_in_folder = spark.createDataFrame(file_data, ["name"])

# Show the DataFrame (this step can be removed)
display(df_files_in_folder)

Step 2 is to list all the files which have already been processed

# This creates a dataframe with the file names of all the files which have already been processed
df_already_processed = spark.sql("SELECT * FROM Lakehouse_bronze.processed_files_log")

# Show the DataFrame (this step can be removed)
display(df_already_processed)

Step 3 is to find out which files in the folder have not been processed yet

# Selecting only the filename column from df_already_processed
df_already_processed_filenames = df_already_processed.select("filename")

# Selecting only the name column from df_files_in_folder
df_files_in_folder_names = df_files_in_folder.select("name")

# Performing subtract operation to find non-matching rows 
# (so only the file names of the files which have not been processed already are kept)
df_to_process = df_files_in_folder_names.subtract(df_already_processed_filenames)

# Showing the resulting DataFrame of files which have not yet been processed (this step can be removed)
display(df_to_process)

Step 4 is the code to process the files which have not already been processed, load the content into the silver table, and create new entries into the log table

from pyspark.sql.functions import current_timestamp, lit, to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, TimestampType
from datetime import datetime

# Loop through the dataframe which consist of the filenames
for row in df_to_process.rdd.collect(): # Collecting to driver (local) as a list of Rows 
    # Extract filename from the current row
    filename = row["name"]
    # Read the current csv file into a dataframe
    df = spark.read.format("csv").option("header", "true").option("inferSchema", "true").load(files_path + filename).select("OrderDateTime", "OrderID", "ProductID", "Quantity")
    # Add filename column to the dataframe
    df = df.withColumn("source_filename", lit(filename))
    tidsstempel = current_timestamp()
    # Add current timestamp column ("source_processedTime") to the dataframe
    df = df.withColumn("source_processedTime", tidsstempel)
    # Append the dataframe to the table in Silver lakehouse
    df.write.format("delta").mode("append").save('<Insert the abfss path to your silver layer Lakehouse table here>')
    # Create a single-row DataFrame with the filename (this will be appended to the log table)
    single_row = [[filename]]
    single_row_schema = StructType([
        StructField("filename", StringType(), False)
    ])
    df_log = spark.createDataFrame(single_row, single_row_schema)
    # Add the processedTime to the single-row DataFrame which will be inserted into the log table
    df_log = df_log.withColumn("processedTime", tidsstempel)
    # Insert the filename and the processedTime into the log table
    df_log.write.mode("append").saveAsTable("processed_files_log")

The raw csv files in Bronze lakehouse look like this:

The destination table in the Silver lakehouse looks like this:

This table contains the data from the csv files, in addition to the source_filename and the time when the source csv file was processed by the Notebook and loaded to the table (source_processedTime).

(If someone notice: The raw csv files just contain dummy data, this is the reason why the OrderID's are repeating between 1-5 for all the csv files.).

If someone have suggestions and/or corrections to this approach and code, please share 😀

Done it. I created an Empty dataframe and then appended into it in the loop you suggested