Can Dataflow Gen2 read from lakehouse "Files" as a folder?

Hi Scott_Powell ,

I wasn't able to figure out loading the files in my folder through Dataflow but ended up using a notebook to ingest the files to a table and went from there. 

Below is the notebook details if you want to give it a shot:

The below loads all files from within a folder and removes the first row in each file (not sure if your files are delmited but below example my files are | delimited), and in the step where you can input the column names you don't have to do this for every column - you can select the columns you want to load you just have to provide the appropiate column # starting from 0

Cell 1

from pyspark.sql.functions import split, row_number, input_file_name
from pyspark.sql import Window

# Read all files from the directory into a DataFrame
df = spark.read.text("Files/Expense/*.txt")

# Add a column for the input file name
df_with_filename = df.withColumn("filename", input_file_name())

# Add a row number to each row, partitioned by the input file name
windowSpec = Window.partitionBy("filename").orderBy("value")
df_with_rownum = df_with_filename.withColumn("rownum", row_number().over(windowSpec))

# Filter out the first row from each file
df_filtered = df_with_rownum.filter(df_with_rownum.rownum > 1)

# Split the value column by the | delimiter
df_split = df_filtered.withColumn("split_values", split(df_filtered["value"], "\|"))

# If you know the number of columns and want to give them names, you can do so
# For example, if there are two columns:
df_final = df_split.select(
    df_split["split_values"].getItem(0).alias("Constant"),
    df_split["split_values"].getItem(1).alias("BatchId"),

)

# Show the resulting DataFrame
df_final.show()

Cell 2

df_final.write.mode("overwrite").format("delta").saveAsTable("TableName")