Forum Discussion
Lakehouse - promote csv files to Tables using For Loop
I'm pretty new to Notebooks ... I trying to promote the 5 csv files in the Lakehouse file directory
Hi Theo86 ,
You can do like this:
#Read Files inside Directory files = mssparkutils.fs.ls("Files/Imported/") # Each item contains file name and path for file in files: # Skip folders inside if not file.isDir: table_name = file.name.replace(".csv","") df = spark.read.option("header", "true").option("inferSchema", "true").csv(file.path) df.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(f"<schema>.{table_name}")You can add a few print statement to keep track of the flow while the code runs. Also you can add try, except block for error-handling.
5 Replies
Hello Theo86
You don't need a For Loop to load all the csv files, you can use wildcards. Please see a code snippet below.
from pyspark.sql.functions import input_file_name, current_date, lit # 1) Read multiple CSVs with wildcard df = (spark.read .option("header", "true") .option("delimiter", ",") .option("quote", '"') .option("escape", '"') .option("inferSchema", "true") .csv("Files/Imported/sen_secondary_need_*.csv")) # 2) Add Bronze lineage/metadata columns df_bronze = (df .withColumn("ingest_date", current_date()) .withColumn("source_file", input_file_name())) # 3) Write to Lakehouse Tables (Delta) spark.sql("CREATE DATABASE IF NOT EXISTS bronze") (df_bronze .write .format("delta") .mode("append") .option("mergeSchema", "true") # tolerates new columns over time .partitionBy("ingest_date") # optional but recommended .saveAsTable("bronze.sen_secondary_need"))This is a code centric approach.
You could also follow a no-code approach using the Lakeouse Explorer.
Right-click the Imported folder > Create table (auto-detect schema) > writes to /Tables.Hi Theo86 ,
You can do like this:
#Read Files inside Directory files = mssparkutils.fs.ls("Files/Imported/") # Each item contains file name and path for file in files: # Skip folders inside if not file.isDir: table_name = file.name.replace(".csv","") df = spark.read.option("header", "true").option("inferSchema", "true").csv(file.path) df.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(f"<schema>.{table_name}")You can add a few print statement to keep track of the flow while the code runs. Also you can add try, except block for error-handling.
Thank you, this is brilliant...😀
The schema on the last line was causing issues copied value and changed the replace for table name.
#Read Files inside Directory files = mssparkutils.fs.ls("Files/Imported/") schema = "bronze" # hard code schema # Each item contains file name and path for file in files: # Skip folders inside if not file.isDir: table_name = file.name.replace("_.csv","") df = spark.read.option("header", "true").option("inferSchema", "true").csv(file.path) df.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(f"{schema}{table_name}")
Hello deborshi_nag
Firstly, thank you for looking and taking the time to write the code, awesome response.
I have moved the wildcard on line 10 so it would pick up all 5 files.
.csv("Files/Imported/sen_*.csv"))However, again I'm new to this (SQL Background but enjoying Notebooks) but it would appear to run through once and finish.
The last line of code would appear to be hardcoded so saves only one file.
======================================================
I have added my crude code just for a complete picture, then I save each df as a delta format table.
.
# Table 1 file_path = 'Files/Imported/sen_age_sex_.csv' df1 = spark.read.load(file_path ,format='csv' ,header=True ,inferSchema=True ) # Table 2 file_path = 'Files/Imported/sen_fsm_eth_lang_new_.csv' df2 = spark.read.load(file_path ,format='csv' ,header=True ,inferSchema=True ) # Table 3 file_path = 'Files/Imported/sen_ncyear_.csv' df3 = spark.read.load(file_path ,format='csv' ,header=True ,inferSchema=True ) # Table 4 file_path = 'Files/Imported/sen_phase_type_.csv' df4 = spark.read.load(file_path ,format='csv' ,header=True ,inferSchema=True ) # Table 5 file_path = 'Files/Imported/sen_secondary_need_.csv' df5 = spark.read.load(file_path ,format='csv' ,header=True ,inferSchema=True )Hi Theo86 I see you have different csv files, each with it's own unique schema. In that case use the code provided by stoic-harsh - that should work!
