Forum Discussion
Scott_Powell
2 years agoAdvocate IV
Help with notebooks / multiple .csv files for complete newbie
Hi, I have a need to ingest a ton of Power BI audit logs into Fabric. The eventual target is a Lakehouse table or tables. I have one log file per day (as csv), about 500 altogether. These files need ...
jwinchell40
2 years agoResolver III
I read a lot of files data that are fed into OneLake via API calls to external systems. Even though my data source are .json files; the code below will work for .csv files as well.
#Imports and Includes to use different functions down the line
from pyspark.sql.functions import *
from pyspark.sql.window import *
from pyspark.sql.types import DateType
from pyspark.sql.functions import to_date,to_timestamp,sequence
from delta.tables import *
##Define a Parameter Cell in the Notebook to be able to dynamically pass a file struct
_year = "9999"
_month = "5"
_day = "2"
##Generating the path to my files dynamically
##Instead of *.json you can use *.csv as this will look for any file that ends .csv for the path specific.
_root = "Files/XCM/TaskSignOff"
_path = "/" + _year + "/" + "/" + _month + "/" + _day + "/*.json"
_full = _root + _path
##If your CSV file has a defined schema, I recommend defining the schema in your Notebook so you can control schema drift. I am calling out the expect elements/headers that will be in the files that I am reading. By setting the schema, if the vendor adds a new field; we do not necessarily bring it in right away.
_schema = StructType(
[
StructField("taskId",StringType()),
StructField("signOffId",StringType()),
StructField("comments",StringType()),
StructField("requiredComments",StringType()),
StructField("requiredDate",StringType()),
StructField("requiredName",StringType()),
StructField("signOffBy",StringType()),
StructField("signOffComments",StringType()),
StructField("signOffDate",StringType()),
StructField("signOffRequired",StringType()),
StructField("signOffName",StringType()),
StructField("statusName",StringType()),
StructField("updatedBy",StringType()),
StructField("updatedOn",StringType())
]
)
##Actual Dataframe ingestion
##Change format form 'json' to 'csv'
##I also append the source path & file name to my column set
_df = spark.read.format('json').schema(_schema).load(_full).select("*","_metadata.file_name","_metadata.file_path")
##Display the results of the read (first 1000 rows)
display(_df)
##Write to Delta Table
##Write Initial Task List To Delta
_df.write.mode("append").format("delta").save("Tables/<Table_Name>")
When you try to write to a file from spark, all of the different work threads have part of the data. So each worker writes their piece of the pie which results in a bunch of files with guids and other oddities for names. In order to write as a single file you would have to call the collect() statement and that would then result in all data getting combined in 1 single worker and then you could write the data. It is a very memory intensive operation.
Hope this helps out a little bit.